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

本文概述

如何使两个线程按顺序执行？
如何使两个线程以指定的方式有序地相交？
在A, B和C都完成同步执行之后执行D
3名选手准备跑步
子线程将结果返回到主线程
总结

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

本文介绍的方法和类是：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）。让我们看一下代码：


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

那就是我们想要的。

怎么了？首先, 我们创建一个由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

在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；否则, 进入计数状态。
在其他线程中调用countDownLatch.countDown()方法, 该方法会将计数值减少一；
当其他线程中的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；否则, 计数器将减少为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放入线程池中以管理执行。

总结

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

本文概述

如何使两个线程按顺序执行？

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

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

3名选手准备跑步

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

总结

相关推荐

评论抢沙发

评论前必须登录！

猜你喜欢

热门标签

回顶部

本文概述

如何使两个线程按顺序执行？

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

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

3名选手准备跑步

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

总结

相关推荐

评论 抢沙发

评论前必须登录！

猜你喜欢

热门标签

回顶部

评论抢沙发