个性化阅读
专注于IT技术分析

如何在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)。让我们看一下代码:


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来实现这种类型的通信。其基本用法是:

  1. 创建一个计数器, 并设置一个初始值CountdownLatch countDownLatch = new CountDownLatch(3;
  2. 在等待线程中调用countDownLatch.await()方法, 并进入等待状态, 直到计数值变为0;否则, 进入计数状态。
  3. 在其他线程中调用countDownLatch.countDown()方法, 该方法会将计数值减少一;
  4. 当其他线程中的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数据结构, 其基本用法是:

  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放入线程池中以管理执行。

总结

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

赞(0)
未经允许不得转载:srcmini » 如何在Java中实现线程间通信

评论 抢沙发

评论前必须登录!