Java中的线程间通信【译】

尽管大多数情况下,每个子线程都只需要完成他们自己的任务,但是有时候我们可能希望多个线程共同协作来执行一个任务,这就涉及到了线程间的通信。

本文要讨论如下的方法和类:thread.join(), object.wait(), object.notify(), CountdownLatch, Cyclicbarrier, FutureTask, Callbale等。

我会用几个例子来解释如何在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()方法允许一个线程在等待另外一个线程完成后继续执行。但是如果我们在D中有序的加入A,B,C,这将使它们按顺序执行,而我们希望三个线程同步地执行。

我们想要实现的目标是:三个线程A,B,C可以在同一时间开始运行,每个线程都会在完成后单独通知D;D直到A,B和C全部执行完后才会执行。所以我们使用CountdownLatch来实现这种类型的通信。

它的基本用法是:

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

所以,在一个线程需要等待其他多个线程的情况下,使用CountDownLatch是个合适的解决方案。

三个运动员准备跑步

三个运动员各自进行准备,然后当他们全部准备好后,同时开始跑步。

这次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对象传递给线程来执行。Runnale的定义如下:

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

总结

多线程在当今编程语言中是一种常见特性,线程间通信,线程同步和线程安全都是非常重要的课题。

发表评论

电子邮件地址不会被公开。 必填项已用*标注