详解Java编程中线程同步以及定时启动线程的方法
使用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
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();
}
}
结果：
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
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();
}
}
结果：
i'm going. he has gone.
4.使用wait()与notify()改写上面的例子
下面的例子通过wait()来取代忙等待机制，当收到通知消息时，notify当前monitor类线程。
wait.java
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
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
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();
}
}
结果：
[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
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();
}
}
结果：
[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();
}
}
}
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();
}
}
结果：
i'm going. he has gone.