第五章 Java线程间的通信

合理的使用Java多线程可以更好地利用服务器资源。一般来讲，线程内部有自己私有的线程上下文，互不干扰。但是当我们需要多个线程之间相互协作的时候，就需要我们掌握Java线程的通信方式。本文将介绍Java线程之间的几种通信原理。

5.1 锁与同步

在Java中，锁的概念都是基于对象的，所以我们又经常称它为对象锁。线程和锁的关系，我们可以用婚姻关系来理解。一个锁同一时间只能被一个线程持有。也就是说，一个锁如果和一个线程“结婚”（持有），那其他线程如果需要得到这个锁，就得等这个线程和这个锁“离婚”（释放）。

在我们的线程之间，有一个同步的概念。什么是同步呢，假如我们现在有2位正在抄暑假作业答案的同学：线程A和线程B。当他们正在抄的时候，老师突然来修改了一些答案，可能A和B最后写出的暑假作业就不一样。我们为了A,B能写出2本相同的暑假作业，我们就需要让老师先修改答案，然后A，B同学再抄。或者A，B同学先抄完，老师再修改答案。这就是线程A，线程B的线程同步。

可以解释为：线程同步是线程之间按照一定的顺序执行。

为了达到线程同步，我们可以使用锁来实现它。

我们先来看看一个无锁的程序：

  1. public class NoneLock {
  3.     static class ThreadA implements Runnable {
  4.         @Override
  5.         public void run() {
  6.             for (int i = 0; i < 100; i++) {
  7.                 System.out.println("Thread A " + i);
  8.             }
  9.         }
  10.     }
  12.     static class ThreadB implements Runnable {
  13.         @Override
  14.         public void run() {
  15.             for (int i = 0; i < 100; i++) {
  16.                 System.out.println("Thread B " + i);
  17.             }
  18.         }
  19.     }
  21.     public static void main(String[] args) {
  22.         new Thread(new ThreadA()).start();
  23.         new Thread(new ThreadB()).start();
  24.     }
  25. }

执行这个程序，你会在控制台看到，线程A和线程B各自独立工作，输出自己的打印值。如下是我的电脑上某一次运行的结果。每一次运行结果都会不一样。

  1. ....
  2. Thread A 48
  3. Thread A 49
  4. Thread B 0
  5. Thread A 50
  6. Thread B 1
  7. Thread A 51
  8. Thread A 52
  9. ....

那我现在有一个需求，我想等A先执行完之后，再由B去执行，怎么办呢？最简单的方式就是使用一个“对象锁”：

  1. public class ObjectLock {
  2.     private static Object lock = new Object();
  4.     static class ThreadA implements Runnable {
  5.         @Override
  6.         public void run() {
  7.             synchronized (lock) {
  8.                 for (int i = 0; i < 100; i++) {
  9.                     System.out.println("Thread A " + i);
  10.                 }
  11.             }
  12.         }
  13.     }
  15.     static class ThreadB implements Runnable {
  16.         @Override
  17.         public void run() {
  18.             synchronized (lock) {
  19.                 for (int i = 0; i < 100; i++) {
  20.                     System.out.println("Thread B " + i);
  21.                 }
  22.             }
  23.         }
  24.     }
  26.     public static void main(String[] args) throws InterruptedException {
  27.         new Thread(new ThreadA()).start();
  28.         Thread.sleep(10);
  29.         new Thread(new ThreadB()).start();
  30.     }
  31. }

这里声明了一个名字为lock的对象锁。我们在ThreadA和ThreadB内需要同步的代码块里，都是用synchronized关键字加上了同一个对象锁lock。

上文我们说到了，根据线程和锁的关系，同一时间只有一个线程持有一个锁，那么线程B就会等线程A执行完成后释放lock，线程B才能获得锁lock。

这里在主线程里使用sleep方法睡眠了10毫秒，是为了防止线程B先得到锁。因为如果同时start，线程A和线程B都是出于就绪状态，操作系统可能会先让B运行。这样就会先输出B的内容，然后B执行完成之后自动释放锁，线程A再执行。

5.2 等待/通知机制

上面一种基于“锁”的方式，线程需要不断地去尝试获得锁，如果失败了，再继续尝试。这可能会耗费服务器资源。

而等待/通知机制是另一种方式。

Java多线程的等待/通知机制是基于Object类的wait()方法和notify(), notifyAll()方法来实现的。

notify()方法会随机叫醒一个正在等待的线程，而notifyAll()会叫醒所有正在等待的线程。

前面我们讲到，一个锁同一时刻只能被一个线程持有。而假如线程A现在持有了一个锁lock并开始执行，它可以使用lock.wait()让自己进入等待状态。这个时候，lock这个锁是被释放了的。

这时，线程B获得了lock这个锁并开始执行，它可以在某一时刻，使用lock.notify()，通知之前持有lock锁并进入等待状态的线程A，说“线程A你不用等了，可以往下执行了”。

需要注意的是，这个时候线程B并没有释放锁lock，除非线程B这个时候使用lock.wait()释放锁，或者线程B执行结束自行释放锁，线程A才能得到lock锁。

我们用代码来实现一下：

  1. public class WaitAndNotify {
  2.     private static Object lock = new Object();
  4.     static class ThreadA implements Runnable {
  5.         @Override
  6.         public void run() {
  7.             synchronized (lock) {
  8.                 for (int i = 0; i < 5; i++) {
  9.                     try {
  10.                         System.out.println("ThreadA: " + i);
  11.                         lock.notify();
  12.                         lock.wait();
  13.                     } catch (InterruptedException e) {
  14.                         e.printStackTrace();
  15.                     }
  16.                 }
  17.                 lock.notify();
  18.             }
  19.         }
  20.     }
  22.     static class ThreadB implements Runnable {
  23.         @Override
  24.         public void run() {
  25.             synchronized (lock) {
  26.                 for (int i = 0; i < 5; i++) {
  27.                     try {
  28.                         System.out.println("ThreadB: " + i);
  29.                         lock.notify();
  30.                         lock.wait();
  31.                     } catch (InterruptedException e) {
  32.                         e.printStackTrace();
  33.                     }
  34.                 }
  35.                 lock.notify();
  36.             }
  37.         }
  38.     }
  40.     public static void main(String[] args) throws InterruptedException {
  41.         new Thread(new ThreadA()).start();
  42.         Thread.sleep(1000);
  43.         new Thread(new ThreadB()).start();
  44.     }
  45. }
  47. // 输出：
  48. ThreadA: 0
  49. ThreadB: 0
  50. ThreadA: 1
  51. ThreadB: 1
  52. ThreadA: 2
  53. ThreadB: 2
  54. ThreadA: 3
  55. ThreadB: 3
  56. ThreadA: 4
  57. ThreadB: 4

在这个Demo里，线程A和线程B首先打印出自己需要的东西，然后使用notify()方法叫醒另一个正在等待的线程，然后自己使用wait()方法陷入等待并释放lock锁。

需要注意的是等待/通知机制使用的是使用同一个对象锁，如果你两个线程使用的是不同的对象锁，那它们之间是不能用等待/通知机制通信的。

5.3 信号量

JDK提供了一个类似于“信号量”功能的类Semaphore。但本文不是要介绍这个类，而是介绍一种基于volatile关键字的自己实现的信号量通信。

后面会有专门的章节介绍volatile关键字，这里只是做一个简单的介绍。

volatile关键字能够保证内存的可见性，如果用volatile关键字声明了一个变量，在一个线程里面改变了这个变量的值，那其它线程是立马可见更改后的值的。

比如我现在有一个需求，我想让线程A输出0，然后线程B输出1，再然后线程A输出2…以此类推。我应该怎样实现呢？

代码：

  1. public class Signal {
  2.     private static volatile int signal = 0;
  4.     static class ThreadA implements Runnable {
  5.         @Override
  6.         public void run() {
  7.             while (signal < 5) {
  8.                 if (signal % 2 == 0) {
  9.                     System.out.println("threadA: " + signal);
  10.                     signal++;
  11.                 }
  12.             }
  13.         }
  14.     }
  16.     static class ThreadB implements Runnable {
  17.         @Override
  18.         public void run() {
  19.             while (signal < 5) {
  20.                 if (signal % 2 == 1) {
  21.                     System.out.println("threadB: " + signal);
  22.                     signal = signal + 1;
  23.                 }
  24.             }
  25.         }
  26.     }
  28.     public static void main(String[] args) throws InterruptedException {
  29.         new Thread(new ThreadA()).start();
  30.         Thread.sleep(1000);
  31.         new Thread(new ThreadB()).start();
  32.     }
  33. }
  35. // 输出：
  36. threadA: 0
  37. threadB: 1
  38. threadA: 2
  39. threadB: 3
  40. threadA: 4

我们可以看到，使用了一个volatile变量signal来实现了“信号量”的模型。这里需要注意的是，volatile变量需要进行原子操作。

需要注意的是，signal++并不是一个原子操作，所以我们在实际开发中，会根据需要使用synchronized给它“上锁”，或者是使用AtomicInteger等原子类。并且上面的程序也并不是线程安全的，因为执行while语句后，可能当前线程就暂停等待时间片了，等线程醒来，可能signal已经大于等于5了。

这种实现方式并不一定高效，本例只是演示信号量

信号量的应用场景：

假如在一个停车场中，车位是我们的公共资源，线程就如同车辆，而看门的管理员就是起的“信号量”的作用。

因为在这种场景下，多个线程（超过2个）需要相互合作，我们用简单的“锁”和“等待通知机制”就不那么方便了。这个时候就可以用到信号量。

其实JDK中提供的很多多线程通信工具类都是基于信号量模型的。我们会在后面第三篇的文章中介绍一些常用的通信工具类。

5.4 管道

管道是基于“管道流”的通信方式。JDK提供了PipedWriter、 PipedReader、 PipedOutputStream、 PipedInputStream。其中，前面两个是基于字符的，后面两个是基于字节流的。

这里的示例代码使用的是基于字符的：

  1. public class Pipe {
  2.     static class ReaderThread implements Runnable {
  3.         private PipedReader reader;
  5.         public ReaderThread(PipedReader reader) {
  6.             this.reader = reader;
  7.         }
  9.         @Override
  10.         public void run() {
  11.             System.out.println("this is reader");
  12.             int receive = 0;
  13.             try {
  14.                 while ((receive = reader.read()) != -1) {
  15.                     System.out.print((char)receive);
  16.                 }
  17.             } catch (IOException e) {
  18.                 e.printStackTrace();
  19.             }
  20.         }
  21.     }
  23.     static class WriterThread implements Runnable {
  25.         private PipedWriter writer;
  27.         public WriterThread(PipedWriter writer) {
  28.             this.writer = writer;
  29.         }
  31.         @Override
  32.         public void run() {
  33.             System.out.println("this is writer");
  34.             int receive = 0;
  35.             try {
  36.                 writer.write("test");
  37.             } catch (IOException e) {
  38.                 e.printStackTrace();
  39.             } finally {
  40.                 try {
  41.                     writer.close();
  42.                 } catch (IOException e) {
  43.                     e.printStackTrace();
  44.                 }
  45.             }
  46.         }
  47.     }
  49.     public static void main(String[] args) throws IOException, InterruptedException {
  50.         PipedWriter writer = new PipedWriter();
  51.         PipedReader reader = new PipedReader();
  52.         writer.connect(reader); // 这里注意一定要连接，才能通信
  54.         new Thread(new ReaderThread(reader)).start();
  55.         Thread.sleep(1000);
  56.         new Thread(new WriterThread(writer)).start();
  57.     }
  58. }
  60. // 输出：
  61. this is reader
  62. this is writer
  63. test

我们通过线程的构造函数，传入了PipedWrite和PipedReader对象。可以简单分析一下这个示例代码的执行流程：

  1. 线程ReaderThread开始执行，
  2. 线程ReaderThread使用管道reader.read()进入”阻塞“，
  3. 线程WriterThread开始执行，
  4. 线程WriterThread用writer.write(“test”)往管道写入字符串，
  5. 线程WriterThread使用writer.close()结束管道写入，并执行完毕，
  6. 线程ReaderThread接受到管道输出的字符串并打印，
  7. 线程ReaderThread执行完毕。

管道通信的应用场景：

这个很好理解。使用管道多半与I/O流相关。当我们一个线程需要先另一个线程发送一个信息（比如字符串）或者文件等等时，就需要使用管道通信了。

5.5 其它通信相关

以上介绍了一些线程间通信的基本原理和方法。除此以外，还有一些与线程通信相关的知识点，这里一并介绍。

5.5.1 join方法

join()方法是Thread类的一个实例方法。它的作用是让当前线程陷入“等待”状态，等join的这个线程执行完成后，再继续执行当前线程。

有时候，主线程创建并启动了子线程，如果子线程中需要进行大量的耗时运算，主线程往往将早于子线程结束之前结束。

如果主线程想等待子线程执行完毕后，获得子线程中的处理完的某个数据，就要用到join方法了。

示例代码：

  1. public class Join {
  2.     static class ThreadA implements Runnable {
  4.         @Override
  5.         public void run() {
  6.             try {
  7.                 System.out.println("我是子线程，我先睡一秒");
  8.                 Thread.sleep(1000);
  9.                 System.out.println("我是子线程，我睡完了一秒");
  10.             } catch (InterruptedException e) {
  11.                 e.printStackTrace();
  12.             }
  13.         }
  14.     }
  16.     public static void main(String[] args) throws InterruptedException {
  17.         Thread thread = new Thread(new ThreadA());
  18.         thread.start();
  19.         thread.join();
  20.         System.out.println("如果不加join方法，我会先被打出来，加了就不一样了");
  21.     }
  22. }

注意join()方法有两个重载方法，一个是join(long)， 一个是join(long, int)。

实际上，通过源码你会发现，join()方法及其重载方法底层都是利用了wait(long)这个方法。

对于join(long, int)，通过查看源码(JDK 1.8)发现，底层并没有精确到纳秒，而是对第二个参数做了简单的判断和处理。

5.5.2 sleep方法

sleep方法是Thread类的一个静态方法。它的作用是让当前线程睡眠一段时间。它有这样两个方法：

  • Thread.sleep(long)
  • Thread.sleep(long, int)

同样，查看源码(JDK 1.8)发现，第二个方法貌似只对第二个参数做了简单的处理，没有精确到纳秒。实际上还是调用的第一个方法。

这里需要强调一下：sleep方法是不会释放当前的锁的，而wait方法会。这也是最常见的一个多线程面试题。

它们还有这些区别：

  • wait可以指定时间，也可以不指定；而sleep必须指定时间。
  • wait释放cpu资源，同时释放锁；sleep释放cpu资源，但是不释放锁，所以易死锁。
  • wait必须放在同步块或同步方法中，而sleep可以在任意位置。

5.5.3 ThreadLocal类

ThreadLocal是一个本地线程副本变量工具类。内部是一个弱引用的Map来维护。这里不详细介绍它的原理，而是只是介绍它的使用，以后有独立章节来介绍ThreadLocal类的原理。

有些朋友称ThreadLocal为线程本地变量或线程本地存储。严格来说，ThreadLocal类并不属于多线程间的通信，而是让每个线程有自己”独立“的变量，线程之间互不影响。它为每个线程都创建一个副本，每个线程可以访问自己内部的副本变量。

ThreadLocal类最常用的就是set方法和get方法。示例代码：

  1. public class ThreadLocalDemo {
  2.     static class ThreadA implements Runnable {
  3.         private ThreadLocal<String> threadLocal;
  5.         public ThreadA(ThreadLocal<String> threadLocal) {
  6.             this.threadLocal = threadLocal;
  7.         }
  9.         @Override
  10.         public void run() {
  11.             threadLocal.set("A");
  12.             try {
  13.                 Thread.sleep(1000);
  14.             } catch (InterruptedException e) {
  15.                 e.printStackTrace();
  16.             }
  17.             System.out.println("ThreadA输出：" + threadLocal.get());
  18.         }
  20.         static class ThreadB implements Runnable {
  21.             private ThreadLocal<String> threadLocal;
  23.             public ThreadB(ThreadLocal<String> threadLocal) {
  24.                 this.threadLocal = threadLocal;
  25.             }
  27.             @Override
  28.             public void run() {
  29.                 threadLocal.set("B");
  30.                 try {
  31.                     Thread.sleep(1000);
  32.                 } catch (InterruptedException e) {
  33.                     e.printStackTrace();
  34.                 }
  35.                 System.out.println("ThreadB输出：" + threadLocal.get());
  36.             }
  37.         }
  39.         public static void main(String[] args) {
  40.             ThreadLocal<String> threadLocal = new ThreadLocal<>();
  41.             new Thread(new ThreadA(threadLocal)).start();
  42.             new Thread(new ThreadB(threadLocal)).start();
  43.         }
  44.     }
  45. }
  47. // 输出：
  48. ThreadA输出：A
  49. ThreadB输出：B

可以看到，虽然两个线程使用的同一个ThreadLocal实例（通过构造方法传入），但是它们各自可以存取自己当前线程的一个值。

那ThreadLocal有什么作用呢？如果只是单纯的想要线程隔离，在每个线程中声明一个私有变量就好了呀，为什么要使用ThreadLocal？

如果开发者希望将类的某个静态变量（user ID或者transaction ID）与线程状态关联，则可以考虑使用ThreadLocal。

最常见的ThreadLocal使用场景为用来解决数据库连接、Session管理等。数据库连接和Session管理涉及多个复杂对象的初始化和关闭。如果在每个线程中声明一些私有变量来进行操作，那这个线程就变得不那么“轻量”了，需要频繁的创建和关闭连接。

5.5.4 InheritableThreadLocal

InheritableThreadLocal类与ThreadLocal类稍有不同，Inheritable是继承的意思。它不仅仅是当前线程可以存取副本值，而且它的子线程也可以存取这个副本值。

参考资料