服务器之家：专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|JAVA教程|ASP教程|

服务器之家 - 编程语言 - JAVA教程 - 详解Java编程中线程同步以及定时启动线程的方法

详解Java编程中线程同步以及定时启动线程的方法

2020-03-20 13:50zhangjunhd JAVA教程

这篇文章主要介绍了详解Java编程中线程同步以及定时启动线程的方法, 讲到了wait()与notify()方法以及阻塞队列等知识,需要的朋友可以参考下

使用wait()与notify()实现线程间协作
1. wait()与notify()/notifyAll()
调用sleep()和yield()的时候锁并没有被释放，而调用wait()将释放锁。这样另一个任务（线程）可以获得当前对象的锁，从而进入它的synchronized方法中。可以通过notify()/notifyAll()，或者时间到期，从wait()中恢复执行。
只能在同步控制方法或同步块中调用wait()、notify()和notifyAll()。如果在非同步的方法里调用这些方法，在运行时会抛出IllegalMonitorStateException异常。
2.模拟单个线程对多个线程的唤醒
模拟线程之间的协作。Game类有2个同步方法prepare()和go()。标志位start用于判断当前线程是否需要wait()。Game类的实例首先启动所有的Athele类实例，使其进入wait()状态，在一段时间后，改变标志位并notifyAll()所有处于wait状态的Athele线程。
Game.java

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
package concurrency;
 
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
 
class Athlete implements Runnable {
  private final int id;
  private Game game;
 
  public Athlete(int id, Game game) {
   this.id = id;
   this.game = game;
  }
 
  public boolean equals(Object o) {
   if (!(o instanceof Athlete))
    return false;
   Athlete athlete = (Athlete) o;
   return id == athlete.id;
  }
 
  public String toString() {
   return "Athlete<" + id + ">";
  }
 
  public int hashCode() {
   return new Integer(id).hashCode();
  }
 
  public void run() {
   try {
    game.prepare(this);
   } catch (InterruptedException e) {
    System.out.println(this + " quit the game");
   }
  }
 }
 
public class Game implements Runnable {
  private Set<Athlete> players = new HashSet<Athlete>();
  private boolean start = false;
 
  public void addPlayer(Athlete one) {
   players.add(one);
  }
 
  public void removePlayer(Athlete one) {
   players.remove(one);
  }
 
  public Collection<Athlete> getPlayers() {
   return Collections.unmodifiableSet(players);
  }
 
  public void prepare(Athlete athlete) throws InterruptedException {
   System.out.println(athlete + " ready!");
   synchronized (this) {
    while (!start)
    wait();
    if (start)
     System.out.println(athlete + " go!");
   }
  }
 
  public synchronized void go() {
   notifyAll();
  }
  
  public void ready() {
   Iterator<Athlete> iter = getPlayers().iterator();
   while (iter.hasNext())
    new Thread(iter.next()).start();
  }
 
  public void run() {
   start = false;
   System.out.println("Ready......");
   System.out.println("Ready......");
   System.out.println("Ready......");
   ready();
   start = true;
   System.out.println("Go!");
   go();
  }
 
  public static void main(String[] args) {
   Game game = new Game();
   for (int i = 0; i < 10; i++)
    game.addPlayer(new Athlete(i, game));
   new Thread(game).start();
  }
}

结果：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
Ready......
Ready......
Ready......
Athlete<0> ready!
Athlete<1> ready!
Athlete<2> ready!
Athlete<3> ready!
Athlete<4> ready!
Athlete<5> ready!
Athlete<6> ready!
Athlete<7> ready!
Athlete<8> ready!
Athlete<9> ready!
Go!
Athlete<9> go!
Athlete<8> go!
Athlete<7> go!
Athlete<6> go!
Athlete<5> go!
Athlete<4> go!
Athlete<3> go!
Athlete<2> go!
Athlete<1> go!
Athlete<0> go!

3.模拟忙等待过程
MyObject类的实例是被观察者，当观察事件发生时，它会通知一个Monitor类的实例（通知的方式是改变一个标志位）。而此Monitor类的实例是通过忙等待来不断的检查标志位是否变化。
BusyWaiting.java

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import java.util.concurrent.TimeUnit;
 
class MyObject implements Runnable {
  private Monitor monitor;
 
  public MyObject(Monitor monitor) {
   this.monitor = monitor;
  }
 
  public void run() {
   try {
    TimeUnit.SECONDS.sleep(3);
    System.out.println("i'm going.");
    monitor.gotMessage();
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
}
 
class Monitor implements Runnable {
  private volatile boolean go = false;
 
  public void gotMessage() throws InterruptedException {
   go = true;
  }
 
  public void watching() {
   while (go == false)
    ;
   System.out.println("He has gone.");
  }
 
  public void run() {
   watching();
  }
}
 
public class BusyWaiting {
  public static void main(String[] args) {
   Monitor monitor = new Monitor();
   MyObject o = new MyObject(monitor);
   new Thread(o).start();
   new Thread(monitor).start();
  }
}

结果：

?
1
2
i'm going.
He has gone.

4.使用wait()与notify()改写上面的例子
下面的例子通过wait()来取代忙等待机制，当收到通知消息时，notify当前Monitor类线程。
Wait.java

?
1
2
3
4
5
6
7
8
9
10
package concurrency.wait;
 
import java.util.concurrent.TimeUnit;
 
class MyObject implements Runnable {
  private Monitor monitor;
 
  public MyObject(Monitor monitor) {
   this.monitor = monitor;
  }

定时启动线程
这里提供两种在指定时间后启动线程的方法。一是通过java.util.concurrent.DelayQueue实现；二是通过java.util.concurrent.ScheduledThreadPoolExecutor实现。
1. java.util.concurrent.DelayQueue
类DelayQueue是一个无界阻塞队列，只有在延迟期满时才能从中提取元素。它接受实现Delayed接口的实例作为元素。
<<interface>>Delayed.java

?
1
2
3
4
5
package java.util.concurrent;
import java.util.*;
public interface Delayed extends Comparable<Delayed> {
  long getDelay(TimeUnit unit);
}

getDelay()返回与此对象相关的剩余延迟时间，以给定的时间单位表示。此接口的实现必须定义一个 compareTo 方法，该方法提供与此接口的 getDelay 方法一致的排序。

DelayQueue队列的头部是延迟期满后保存时间最长的 Delayed 元素。当一个元素的getDelay(TimeUnit.NANOSECONDS) 方法返回一个小于等于 0 的值时，将发生到期。
2.设计带有时间延迟特性的队列
类DelayedTasker维护一个DelayQueue<DelayedTask> queue，其中DelayedTask实现了Delayed接口，并由一个内部类定义。外部类和内部类都实现Runnable接口，对于外部类来说，它的run方法是按定义的时间先后取出队列中的任务，而这些任务即内部类的实例，内部类的run方法定义每个线程具体逻辑。

这个设计的实质是定义了一个具有时间特性的线程任务列表，而且该列表可以是任意长度的。每次添加任务时指定启动时间即可。
DelayedTasker.java

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
package com.zj.timedtask;
 
import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
 
import java.util.Collection;
import java.util.Collections;
import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
 
public class DelayedTasker implements Runnable {
  DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>();
 
  public void addTask(DelayedTask e) {
    queue.put(e);
  }
 
  public void removeTask() {
    queue.poll();
  }
 
  public Collection<DelayedTask> getAllTasks() {
    return Collections.unmodifiableCollection(queue);
  }
 
  public int getTaskQuantity() {
    return queue.size();
  }
 
  public void run() {
    while (!queue.isEmpty())
      try {
       queue.take().run();
      } catch (InterruptedException e) {
       System.out.println("Interrupted");
      }
    System.out.println("Finished DelayedTask");
  }
 
  public static class DelayedTask implements Delayed, Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private final int delta;
    private final long trigger;
 
    public DelayedTask(int delayInSeconds) {
      delta = delayInSeconds;
      trigger = System.nanoTime() + NANOSECONDS.convert(delta, SECONDS);
    }
 
    public long getDelay(TimeUnit unit) {
      return unit.convert(trigger - System.nanoTime(), NANOSECONDS);
    }
 
    public int compareTo(Delayed arg) {
      DelayedTask that = (DelayedTask) arg;
      if (trigger < that.trigger)
       return -1;
      if (trigger > that.trigger)
       return 1;
      return 0;
    }
 
    public void run() {
      //run all that you want to do
      System.out.println(this);
    }
 
    public String toString() {
      return "[" + delta + "s]" + "Task" + id;
    }
  }
 
  public static void main(String[] args) {
    Random rand = new Random();
    ExecutorService exec = Executors.newCachedThreadPool();
    DelayedTasker tasker = new DelayedTasker();
    for (int i = 0; i < 10; i++)
      tasker.addTask(new DelayedTask(rand.nextInt(5)));
    exec.execute(tasker);
    exec.shutdown();
  }
}

结果：

?
1
2
3
4
5
6
7
8
9
10
11
[0s]Task 1
[0s]Task 2
[0s]Task 3
[1s]Task 6
[2s]Task 5
[3s]Task 8
[4s]Task 0
[4s]Task 4
[4s]Task 7
[4s]Task 9
Finished DelayedTask

3. java.util.concurrent.ScheduledThreadPoolExecutor
该类可以另行安排在给定的延迟后运行任务（线程），或者定期（重复）执行任务。在构造子中需要知道线程池的大小。最主要的方法是：

[1] schedule
public ScheduledFuture<?> schedule(Runnable command, long delay,TimeUnit unit)
创建并执行在给定延迟后启用的一次性操作。
指定者：
-接口 ScheduledExecutorService 中的 schedule;
参数：
-command - 要执行的任务 ;
-delay - 从现在开始延迟执行的时间 ;
-unit - 延迟参数的时间单位 ;
返回：
-表示挂起任务完成的 ScheduledFuture，并且其 get() 方法在完成后将返回 null。
 
[2] scheduleAtFixedRate
public ScheduledFuture<?> scheduleAtFixedRate(
Runnable command,long initialDelay,long period,TimeUnit unit)
创建并执行一个在给定初始延迟后首次启用的定期操作，后续操作具有给定的周期；也就是将在 initialDelay 后开始执行，然后在 initialDelay+period 后执行，接着在 initialDelay + 2 * period 后执行，依此类推。如果任务的任何一个执行遇到异常，则后续执行都会被取消。否则，只能通过执行程序的取消或终止方法来终止该任务。如果此任务的任何一个执行要花费比其周期更长的时间，则将推迟后续执行，但不会同时执行。
指定者：
-接口 ScheduledExecutorService 中的 scheduleAtFixedRate;
参数：
-command - 要执行的任务 ;
-initialDelay - 首次执行的延迟时间 ;
-period - 连续执行之间的周期 ;
-unit - initialDelay 和 period 参数的时间单位 ;
返回：
-表示挂起任务完成的 ScheduledFuture，并且其 get() 方法在取消后将抛出异常。
4.设计带有时间延迟特性的线程执行者
类ScheduleTasked关联一个ScheduledThreadPoolExcutor，可以指定线程池的大小。通过schedule方法知道线程及延迟的时间，通过shutdown方法关闭线程池。对于具体任务（线程）的逻辑具有一定的灵活性（相比前一中设计，前一种设计必须事先定义线程的逻辑，但可以通过继承或装饰修改线程具体逻辑设计）。
ScheduleTasker.java

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
package com.zj.timedtask;
 
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
public class ScheduleTasker {
  private int corePoolSize = 10;
  ScheduledThreadPoolExecutor scheduler;
 
  public ScheduleTasker() {
    scheduler = new ScheduledThreadPoolExecutor(corePoolSize);
  }
 
  public ScheduleTasker(int quantity) {
    corePoolSize = quantity;
    scheduler = new ScheduledThreadPoolExecutor(corePoolSize);
  }
 
  public void schedule(Runnable event, long delay) {
    scheduler.schedule(event, delay, TimeUnit.SECONDS);
  }
 
  public void shutdown() {
    scheduler.shutdown();
  }
 
  public static void main(String[] args) {
    ScheduleTasker tasker = new ScheduleTasker();
    tasker.schedule(new Runnable() {
      public void run() {
       System.out.println("[1s]Task 1");
      }
    }, 1);
    tasker.schedule(new Runnable() {
      public void run() {
       System.out.println("[2s]Task 2");
      }
    }, 2);
    tasker.schedule(new Runnable() {
      public void run() {
       System.out.println("[4s]Task 3");
      }
    }, 4);
    tasker.schedule(new Runnable() {
      public void run() {
       System.out.println("[10s]Task 4");
      }
    }, 10);
 
    tasker.shutdown();
  }
}

结果：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
[1s]Task 1
[2s]Task 2
[4s]Task 3
[10s]Task 4
  public void run() {
   try {
    TimeUnit.SECONDS.sleep(3);
    System.out.println("i'm going.");
    monitor.gotMessage();
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
}
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class Monitor implements Runnable {
  private volatile boolean go = false;
 
  public synchronized void gotMessage() throws InterruptedException {
   go = true;
   notify();
  }
 
  public synchronized void watching() throws InterruptedException {
   while (go == false)
    wait();
   System.out.println("He has gone.");
  }
 
  public void run() {
   try {
    watching();
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
}
 
public class Wait {
  public static void main(String[] args) {
   Monitor monitor = new Monitor();
   MyObject o = new MyObject(monitor);
   new Thread(o).start();
   new Thread(monitor).start();
  }
}

结果：

?
1
2
i'm going.
He has gone.

 

  • Java编程
  • 线程同步
  • 启动线程

    • 延伸 · 阅读

    精彩推荐
    最近更新
    编辑推荐
    阅读排行
    热门标签
    501 　 2000 　 1433 　 波形 　 JMagick 　 Sedna 　 验证算法 　 存取图片 　 Brute 　 iText 　 一个月 　 多少天 　 多少周 　 wav 　 Insertion 　 Force 　 三角数字 　 Fast 　 Counting 　 考拉兹猜想 　 Collatz 　 Conjecture 　 公式运算 　 汉诺塔问题 　 Hanoi 　 Selection 　 缩放 　 补白 　 首拼 　 SGIP 　
    © 2019-2020 服务器之家 版权所有 www.zzvips.com 关于我们联系我们版权申明网站地图