如何在Java中实现线程间通信?

参考

How To Implement Inter-thread Communication In Java

如何在Java中实现线程间通信?

尽管通常每个子线程只需要完成自己的任务,但是有时我们可能希望多个线程协同工作来完成一项任务,这涉及线程间的通信。

本文介绍的方法和类是:thread.join(),object.wait(),object.notify(),CountdownLatch,CyclicBarrier,FutureTask,Callable等。

这 是本文涵盖的代码

我将使用几个示例来说明如何在Java中实现线程间通信。

如何使两个线程依次执行?如何使两个线程以指定的方式有序相交?有四个线程:A,B,C和D(在A,B和C都完成执行并且A,B和C必须同步执行之前,不会执行D)。三名运动员准备分开,然后在他们各自准备就绪后同时开始跑步。子线程完成任务后,它将结果返回给主线程。

如何使两个线程依次执行?

假设有两个线程:线程A和线程B。这两个线程都可以依次打印三个数字(1-3)。让我们看一下代码:

<code>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(); }/<code>

printNumber(String)的实现如下,它用于按顺序打印三个数字1、2和3:

<code>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); } }/<code>

我们得到的结果是:

<code>B print: 1 A print: 1 B print: 2 A print: 2 B print: 3 A print: 3 /<code>

您可以看到A和B同时打印数字。
那么,如果我们希望B在A打印完之后开始打印呢?我们可以使用thread.join()方法,代码如下:

<code>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(); }/<code>

现在获得的结果是:

<code>B starts waiting for A A print: 1 A print: 2 A print: 3 B print: 1 B print: 2 B print: 3 /<code>

因此,我们可以看到A.join()方法将使B等待直到A完成打印。

如何使两个线程以指定的方式有序相交?

那么,如果现在我们希望B在A打印完1之后立即开始打印1,2,3,然后A继续打印2,3呢?显然,我们需要更多细粒度的锁来控制执行顺序。

在这里,我们可以利用object.wait()和object.notify()方法的优势。代码如下:

<code>/** * 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(); }/<code>

结果如下:

<code>A 1 A waiting… B 1 B 2 B 3 A 2 A 3 /<code>

那就是我们想要的。

做了什么?

首先,我们创建一个由A和B共享的对象锁:lock = new Object();当A获得锁时,它首先打印1,然后调用lock.wait()方法,该方法将使其进入等待状态,然后移交对锁的控制。直到A调用lock.wait()方法释放控制并且B获得锁,B才会执行。B在获得锁后打印1,2,3,然后调用lock.notify()方法唤醒正在等待的A;唤醒后,A将继续打印其余的2、3。

我将日志添加到上述代码中,以使其更易于理解。

<code>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();/<code>

结果如下:

<code>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 /<code>

在A,B和C都完成同步执行之后执行D

前面介绍的方法thread.join()允许一个线程在等待另一个线程完成运行之后继续执行。但是,如果我们将A,B和C顺序连接到D线程中,它将使A,B和C依次执行,而我们希望它们三个同步运行。

我们要实现的目标是:三个线程A,B和C可以同时开始运行,并且每个线程在独立运行完成后都会通知D;在A,B和C全部完成运行之后,D才开始运行。因此,我们使用CountdownLatch来实现这种类型的通信。其基本用法是:

创建一个计数器,并设置一个初始值CountdownLatch countDownLatch = new CountDownLatch(3);在等待线程中调用countDownLatch.await()方法,并进入等待状态,直到计数值变为0;否则,计数值为0。在其他线程中调用countDownLatch.countDown()方法,该方法会将计数值减少一;当其他线程中的countDown()方法将计数值设为0时,等待线程中的countDownLatch.await()方法将立即退出并继续执行以下代码。

实现代码如下:

<code>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(); } }/<code>

结果如下:

<code>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 /<code>

实际上,CountDownLatch本身就是一个倒数计数器,我们将初始计数值设置为3。当D运行时,它首先调用countDownLatch.await()方法来检查计数器值是否为0,并且它将保持等待状态。 如果该值不为0。A,B和C分别运行完后,将分别使用countDownLatch.countDown()方法将倒数计数器减1。 当它们全部三个完成运行时,计数器将减少为0;否则,计数器将减少为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()方法时,这意味着这些线程已经准备就绪,那么这些线程将开始继续同时执行。

实现代码如下。想象一下,有三个跑步者需要同时开始跑步,因此他们需要等待其他跑步者准备就绪。

<code>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(); } }/<code>

结果如下:

<code>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 /<code>

子线程将结果返回到主线程

在实际开发中,通常我们需要创建子线程来执行一些耗时的任务,然后将执行结果传递回主线程。那么如何用Java实现呢?
因此,通常,在创建线程时,我们会将Runnable对象传递给Thread以便执行。 Runnable的定义如下:

<code>public interface Runnable { public abstract void run(); }/<code>

您可以看到run()方法执行后不会返回任何结果。如果要返回结果怎么办?在这里,您可以使用另一个类似的接口类Callable:

<code>@FunctionalInterface public interface Callable { /** * 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; }/<code>

可以看出,Callable的最大区别在于它返回了泛型。

因此,下一个问题是,如何将子线程的结果传递回去? Java有一个FutureTask类,可以与Callable一起使用,但是请注意,用于获取结果的get方法将阻塞主线程。

例如,我们希望子线程计算从1到100的总和,然后将结果返回给主线程。

<code>private static void doTaskWithResultInWorker() { Callable callable = new Callable() { @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 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(); } }/<code>

结果如下:

<code>Before futureTask.get() Task starts Task finished and return result Result: 5050 After futureTask.get() /<code>

可以看出,当主线程调用futureTask.get()方法时,它将阻塞主线程。 然后Callable开始在内部执行并返回操作结果; 然后futureTask.get()获取结果,主线程恢复运行。

在这里,我们可以了解到FutureTask和Callable可以直接在主线程中获取子线程的结果,但是它们将阻塞主线程。 当然,如果您不想阻塞主线程,则可以考虑使用ExecutorService将FutureTask放入线程池中以管理执行。

总结

多线程是现代语言的常见功能,线程间通信,线程同步和线程安全是非常重要的主题。