翻译自TutorialDocs
虽然通常每个子线程只需要完成自己的任务,但有时我们可能希望多个线程一起工作来完成一个任务,这就涉及到线程间的通信。
本文会涉及到的方法和类有:thread.join(), object.wait(), object.notify(), CountdownLatch, CyclicBarrier, FutureTask, Callable
等。
本文涉及到的代码可在这里找到。
我将使用几个示例来解释如何在Java中实现线程间通信。
- 如何使两个线程按顺序执行?
- 如何使两个线程以指定的方式有序交叉执行?
- 有四个线程:A,B,C和D(直到A,B和C都执行完毕才会执行D,并且A,B和C将同步执行)。
- 三名运动员分别准备,每个运动员准备好后同时开始跑步。
- 子线程完成任务后,将结果返回给主线程。
如何使两个线程按顺序执行?
假设有两个线程:线程A和线程B.两个线程都可以按顺序打印三个数字(1-3)。我们来看看代码:
private static void demo1() {
Thread A = new Thread(new Runnable() {
@Override
public void run() {
printNumber("A");
}
});
Thread B = new Thread(new Runnable() {
@Override
public void run() {
printNumber("B");
}
});
A.start();
B.start();
}
printNumber(String)
方法的实现如下,用于按顺序打印三个数字1,2和3:
private static void printNumber(String threadName) {
int i=0;
while (i++ < 3) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(threadName + "print:" + i);
}
}
我们得到如下结果:
B print: 1
A print: 1
B print: 2
A print: 2
B print: 3
A print: 3
你可以看到A和B同时打印数字。
那么,如果我们希望B在A打印完毕后开始打印该怎么办?我们可以使用thread.join()
方法,代码如下:
private static void demo2() {
Thread A = new Thread(new Runnable() {
@Override
public void run() {
printNumber("A");
}
});
Thread B = new Thread(new Runnable() {
@Override
public void run() {
System.out.println("B starts waiting for A");
try {
A.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
printNumber("B");
}
});
B.start();
A.start();
}
现在得到的结果是:
B starts waiting for A
A print: 1
A print: 2
A print: 3
B print: 1
B print: 2
B print: 3
所以我们可以看到A.join()
方法会让B等到A完成打印才开始。
如何使两个线程以指定的方式有序交叉执行?
那么如果现在我们希望B在A打印1之后开始打印1,2,3,那么A继续打印2,3?显然,我们需要更细粒度的锁来控制执行顺序。
这里,我们可以利用object.wait()
和object.notify()
方法。代码如下:
/**
* A 1, B 1, B 2, B 3, A 2, A 3
*/
private static void demo3() {
Object lock = new Object();
Thread A = new Thread(new Runnable() {
@Override
public void run() {
synchronized (lock) {
System.out.println("A 1");
try {
lock.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("A 2");
System.out.println("A 3");
}
}
});
Thread B = new Thread(new Runnable() {
@Override
public void run() {
synchronized (lock) {
System.out.println("B 1");
System.out.println("B 2");
System.out.println("B 3");
lock.notify();
}
}
});
A.start();
B.start();
}
结果如下:
A 1
B 1
B 2
B 3
A 2
A 3
这就得到了我们想要的。
发生了什么?
- 首先,我们创建一个由A和B共享的对象锁:
lock = new Object()
;- 当A获得锁时,它首先打印1,然后调用
lock.wait()
方法,使其进入等待状态,然后交出锁的控制权;- 在A调用
lock.wait()
方法释放控制权并且B获得锁之前,B不会被执行;- 获取锁后B打印1,2,3,然后调用
lock.notify()
方法唤醒正在等待的A线程;- A唤醒后继续打印剩余的2,3。
给上面的代码添加日志,以便更容易理解。
private static void demo3() {
Object lock = new Object();
Thread A = new Thread(new Runnable() {
@Override
public void run() {
System.out.println("INFO: A is waiting for the lock");
synchronized (lock) {
System.out.println("INFO: A got the lock");
System.out.println("A 1");
try {
System.out.println("INFO: A is ready to enter the wait state, giving up control of the lock");
lock.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("INFO: B wakes up A, and A regains the lock");
System.out.println("A 2");
System.out.println("A 3");
}
}
});
Thread B = new Thread(new Runnable() {
@Override
public void run() {
System.out.println("INFO: B is waiting for the lock");
synchronized (lock) {
System.out.println("INFO: B got the lock");
System.out.println("B 1");
System.out.println("B 2");
System.out.println("B 3");
System.out.println("INFO: B ends printing, and calling the notify method");
lock.notify();
}
}
});
A.start();
B.start();
结果如下:
INFO: A is waiting for the lock
INFO: A got the lock
A 1
INFO: A is ready to enter the wait state, giving up control of the lock
INFO: B is waiting for the lock
INFO: B got the lock
B 1
B 2
B 3
INFO: B ends printing, and calling the notify method
INFO: B wakes up A, and A regains the lock
A 2
A 3
D在A,B,C都同步执行完后才执行
前面介绍的thread.join()
方法允许一个线程在等待另一个线程执行完后继续执行。如果我们将A,B,C有序的join
在D之前,会顺序执行A,B,C,再执行D;但如果我们希望A,B,C同步执行完再执行D该怎么做呢?
我们想要达到的目标是:三个线程A,B和C同时开始执行,每个线程在独立执行完后通知D;在A,B和C全部执行完之前,D不会开始运行。
这里我们用CountdownLatch
来实现这种通信。基本用法如下:
- 创建一个计数器,并设置一个初始值:
CountdownLatch countDownLatch = new CountDownLatch(3)
; - 在等待线程中调用
countDownLatch.await()
方法并进入等待状态,直到计数器变为0; - 在其他线程中调用
countDownLatch.countDown()
方法,该方法将计数器的值减1; - 当其他线程中的
countDown()
方法将计数值变为0时,等待线程中的countDownLatch.await()
方法将立即退出,继续执行下面的代码。
实现代码如下:
private static void runDAfterABC() {
int worker = 3;
CountDownLatch countDownLatch = new CountDownLatch(worker);
new Thread(new Runnable() {
@Override
public void run() {
System.out.println("D is waiting for other three threads");
try {
countDownLatch.await();
System.out.println("All done, D starts working");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
for (char threadName='A'; threadName <= 'C'; threadName++) {
final String tN = String.valueOf(threadName);
new Thread(new Runnable() {
@Override
public void run() {
System.out.println(tN + "is working");
try {
Thread.sleep(100);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(tN + "finished");
countDownLatch.countDown();
}
}).start();
}
}
结果如下:
D is waiting for other three threads
A is working
B is working
C is working
A finished
C finished
B finished
All done, D starts working
实际上,CountDownLatch
本身就是倒计时计数器,我们将初始计数值设置为3。当D运行时,它首先调用countDownLatch.await()
方法来检查计数器值是否为0,如果值不为0,它将保持等待状态。A,B和C各自运行完成后将使用countDownLatch.countDown()
方法把倒计时计数器减1。当所有这三个线程运行完成时,计数器将减少到0;然后,D的await()
方法将被触发来结束A,B和C线程,D将开始继续执行。
因此,CountDownLatch
适用于一个线程需要等待多个线程的情况。
3名跑步者准备跑步
三名运动员分别准备,每个运动员准备好后同时开始跑步。
这次,三个线程A,B和C中的每一个都需要单独准备,它们在三个线程都准备就绪后开始同时运行。我们该如何实现这一目标?
上面的CountDownLatch
可用于倒计时,但是当计数完成时,只有一个线程的await()
方法将获得响应,无法同时触发多个线程。
为了实现线程相互等待的效果,我们可以使用CyclicBarrier
数据结构,基本用法如下:
- 首先创建一个公共对象
CyclicBarrier
,并设置同时等待的线程数:CyclicBarrier cyclicBarrier = new CyclicBarrier(3)
; - 这些线程开始同时准备。某个线程准备好后,需要等待其他线程完成准备,所以调用
cyclicBarrier.await()
方法等待其他线程; - 当需要同时等待的指定线程全部调用
cyclicBarrier.await()
方法时,意味着这些线程都已准备就绪,然后这些线程将开始同时继续执行。
实现代码如下。想象一下,有三个跑步者需要同时开始起跑,他们需要等待其他人,直到他们都准备好了。
private static void runABCWhenAllReady() {
int runner = 3;
CyclicBarrier cyclicBarrier = new CyclicBarrier(runner);
final Random random = new Random();
for (char runnerName='A'; runnerName <= 'C'; runnerName++) {
final String rN = String.valueOf(runnerName);
new Thread(new Runnable() {
@Override
public void run() {
long prepareTime = random.nextInt(10000) + 100;
System.out.println(rN + "is preparing for time:" + prepareTime);
try {
Thread.sleep(prepareTime);
} catch (Exception e) {
e.printStackTrace();
}
try {
System.out.println(rN + "is prepared, waiting for others");
cyclicBarrier.await(); // The current runner is ready, waiting for others to be ready
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(rN + "starts running"); // All the runners are ready to start running together
}
}).start();
}
}
结果如下:
A is preparing for time: 4131
B is preparing for time: 6349
C is preparing for time: 8206
A is prepared, waiting for others
B is prepared, waiting for others
C is prepared, waiting for others
C starts running
A starts running
B starts running
子线程将结果返回到主线程
在实际开发中,我们经常需要创建子线程来执行一些耗时的任务,然后将执行结果传递回主线程。那么如何在Java中实现呢?
通常,在创建线程时,我们会将Runnable
对象传递给Thread
进行执行。 Runnable
的定义如下:
public interface Runnable {
public abstract void run();
}
可以看到run()
方法在执行后不返回任何结果。如果想返回结果怎么办?在这里可以使用另一个类似的接口类Callable
:
@FunctionalInterface
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}
可以看出,Callable
的最大区别在于它返回了泛型。
接下来的问题是,如何将子线程的结果传回去?Java有一个类:FutureTask
,可以与Callable
配合使用。但请注意,用于获取结果的get
方法将阻塞主线程。
举个例子,我们希望子线程计算从1到100的总和,并将结果返回给主线程。
private static void doTaskWithResultInWorker() {
Callable<Integer> callable = new Callable<Integer>() {
@Override
public Integer call() throws Exception {
System.out.println("Task starts");
Thread.sleep(1000);
int result = 0;
for (int i=0; i<=100; i++) {
result += i;
}
System.out.println("Task finished and return result");
return result;
}
};
FutureTask<Integer> futureTask = new FutureTask<>(callable);
new Thread(futureTask).start();
try {
System.out.println("Before futureTask.get()");
System.out.println("Result:" + futureTask.get());
System.out.println("After futureTask.get()");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
结果如下:
Before futureTask.get()
Task starts
Task finished and return result
Result: 5050
After futureTask.get()
可以看出,当主线程调用futureTask.get()
方法时,它会阻塞主线程;然后Callable
开始在内部执行并返回操作的结果;然后futureTask.get()
获取结果,主线程恢复运行。
在这里我们可以了解到FutureTask
和Callable
可以直接在主线程中获得子线程的结果,但是它们会阻塞主线程。当然,如果你不想阻塞主线程,可以考虑使用ExecutorService
将FutureTask
放入线程池来管理执行。
总结
多线程是现代语言的常见特性,线程间通信,线程同步和线程安全是非常重要的主题。