发布于 5年前

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

虽然通常每个子线程只需要完成自己的任务，但有时我们可能希望多个线程一起工作来完成一个任务，这涉及到线程间的通信。

本文中涉及的方法和类是：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
A waiting…

B 1
B 2
B 3
A 2
A 3

这就是我们想要的。

发生了什么？

  1. 首先，我们创建一个由A和B共享的对象锁： lock = new Object();
  2. 当A获得锁定时，它首先打印1，然后调用lock.wait()方法使其进入等待状态，然后移交锁定控制;
  3. 在A调用lock.wait()释放控制的方法并且B获得锁定之前，B不会被执行;
  4. B在获得锁定后打印1,2,3，然后调用lock.notify()方法唤醒正在等待的A;
  5. 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有序地连接到D线程中，它将依次执行A，B和C，而我们希望它们三个同步运行。

我们要实现的目标是：三个线程A，B和C可以同时开始运行，每个线程在完成独立运行后都会通知D; 在A，B和C全部完成运行之前，D不会开始运行。所以我们 CountdownLatch 用来实现这种类型的通信。它的基本用法是：

  1. 创建一个计数器，并设置一个初始值， CountdownLatch countDownLatch = new CountDownLatch(3; 2.countDownLatch.await() 在等待线程中调用该方法并进入等待状态，直到计数值变为0;
  2. countDownLatch.countDown() 在其他线程中调用该方法，该方法将计数值减1;
  3. 当 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; 然后，await() D 的方法将被触发结束A，B和C，D将开始继续执行。

因此，CountDownLatch 适用于一个线程需要等待多个线程的情况。

3名跑步者准备跑步

三个跑步者准备分开，然后他们在每个准备好之后开始同时跑步。

这次，三个线程A，B和C中的每一个都需要单独准备，然后它们在三个线程都准备就绪后开始同时运行。我们该如何实现这一目标？

在 CountDownLatch上面可以用倒计时，但在完成计数的时候，只有一个线程的一个 await() 方法会得到响应，所以多线程不能在同一时间被触发。

为了实现线程相互等待的效果，我们可以使用 CyclicBarrier 数据结构，其基本用法是：

  1. 首先创建一个公共对象 CyclicBarrier，并设置同时等待的线程数，CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
  2. 这些线程开始同时准备。准备好之后，他们需要等待其他人完成准备，所以要求 cyclicBarrier.await() 方法等待其他人;
  3. 当需要等待的指定线程全部调用该 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 线程池来管理执行。

摘要

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

译文地址 https://www.tutorialdocs.com/article/java-inter-thread-communication.html

上一篇： mysql查询时如何格式化时间 下一篇： 深入学习7个PHP预定义接口

相关文章

纯净的上网主页

热门推荐

©2020 edoou.com   京ICP备16001874号-3