Java中线程之间的通信方式

文章目录

1 volatile
2 synchronized
3 等待/通知机制——wait(), notify()
- (1) 等待/通知的相关方法：
- (2) 等待/通知的经典范式
- (3) 循环队列：多生产者多消费者模型
4 管道输入/输出流
5 Thread.join()
6 ThreadLocal——线程变量

线程同步：即当有一个线程在对内存进行操作时，其他线程都不可以对这个内存地址进行操作，直到该线程完成操作，其他线程才能对该内存地址进行操作。同步的话，仅仅传递的是控制信息，就是我什么时候运行结束，你什么时候可以来。

对于线程间通信来说，线程间同步可以归纳为线程间通信的一个子集，对于线程通信指的是两个线程之间可以交换一些实时的数据信息，而线程同步只交换一些控制信息。

1 volatile

关键字volatile可以用来修饰字段（成员变量），就是告知程序任何对该变量的访问均需要从共享内存中获取，而对它的改变必须同步刷新回共享内存，它能保证所有线程对变量访问的可见性。

2 synchronized

关键字synchronized可以修饰方法或者以同步块的形式来进行使用，它主要确保多个线程在同一个时刻，只能有一个线程处于方法或者同步块中，它保证了线程对变量访问的可见性和排他性。

任意一个对象都拥有自己的监视器，当这个对象由同步块或者这个对象的同步方法调用时，执行方法的线程必须先获取到该对象的监视器才能进入同步块或者同步方法，而没有获取到监视器（执行该方法）的线程将会被阻塞在同步块和同步方法的入口处，进入BLOCKED状态。

对象、对象的监视器、同步队列和执行线程之间的关系：

任意线程对Object（Object由synchronized保护）的访问，首先要获得Object的监视器。如果获取失败，线程进入同步队列，线程状态变为BLOCKED。当访问Object的前驱（获得了锁的线程）释放了锁，则该释放操作唤醒阻塞在同步队列中的线程，使其重新尝试对监视器的获取。

synchronized线程间通信举例：

class Target{synchronized public void methodA(){System.out.println("hello methodA");}synchronized  public void methodB(){System.out.println("hello methodB");}
}class A extends Thread{Target target;public A(Target target){this.target = target;}@Overridepublic void run() {target.methodA();}
}class B extends Thread{Target target;public B(Target target){this.target = target;}@Overridepublic void run() {target.methodB();}
}
public class Main {public static void main(String[] args) {Target target = new Target(); //A B线程持有相同的对象形成互斥A a = new A(target); //创建一个A线程B b = new B(target); //创建一个B线程a.start();b.start();}
}

本质上就是“共享内存”式的通信。A, B两个个线程需要访问同一个共享变量target，谁先拿到了锁，谁就可以先执行，而令一个线程需要等待。这样，线程A和线程B就实现了通信。

3 等待/通知机制——wait(), notify()

等待/通知的相关方法是任意Java对象都具备的，因为这些方法被定义在所有对象的超类java.lang.Object上。

(1) 等待/通知的相关方法：

等待/通知机制，是指一个线程A调用了对象O的wait()方法进入等待状态，而另一个线程B调用了对象O的notify()或者notifyAll()方法，线程A收到通知后从对象O的wait()方法返回，进而执行后续操作。

wait(),notify的注意事项：

1）使用wait()、notify()和notifyAll()时需要先对调用对象加锁。

2）调用wait()方法后，线程状态由RUNNING变为WAITING，并将当前线程放置到对象的等待队列。

3）notify()或notifyAll()方法调用后，等待线程依旧不会从wait()返回，需要调用notify()或notifAll()的线程释放锁之后，等待线程才有机会从wait()返回。

4）notify()方法将等待队列中的一个等待线程从等待队列中移到同步队列中，而notifyAll()方法则是将等待队列中所有的线程全部移到同步队列，被移动的线程状态由WAITING变为BLOCKED。

5）从wait()方法返回的前提是获得了调用对象的锁。

注：
等待/通知机制依托于同步机制，其目的就是确保等待线程从wait()方法返回时能够感知到通知线程对变量做出的修改。

在synchronized修饰的同步方法或者修饰的同步代码块中使用Object类提供的wait(),notify()和notifyAll()3个方法进行线程通信。

(2) 等待/通知的经典范式

等待方遵循如下原则。
1）获取对象的锁。
2）如果条件不满足，那么调用对象的wait()方法，被通知后仍要检查条件。
3）条件满足则执行对应的逻辑。

synchronized(对象){while(条件不满足){对象.wait();}//处理逻辑部分
}

通知方遵循如下原则。
1）获得对象的锁。
2）改变条件。
3）通知所有等待在对象上的线程。

synchronized(对象){改变条件；对象.notifyAll();
}

(3) 循环队列：多生产者多消费者模型

import java.util.Scanner;
public class MyBlockingArrayQueue {int[] array = new int[3];  // 下标处的数据可能出现生产者和消费者修改同一处的情况int front = 0;  // 只有消费者修改 frontint rear = 0;   // 只有生产者修改 rearint size = 0;   // size 是生产者消费者都会修改的// 生产者才会调用 putsynchronized void put(int value) throws InterruptedException {// 考虑满的情况while (size == array.length) {// 队列已满，等待消费线程去取wait(); }// 通过 while 循环，保证了，走到这里时，size 一定不等于 array.lengtharray[rear] = value;rear = (rear + 1) % array.length;size++;  // 我们需要保障的是 size++ 的原子性，所以 volatile 无法解决System.out.println(size); // 1 - 10notifyAll();   // 我们以为我们唤醒的是消费者线程，但实际可能唤醒了生产者线程}// 调用 take 的一定是消费者synchronized int take() throws InterruptedException {// 考虑空的情况while (size == 0) {// 空的，等待生产者线程工作wait();}int value = array[front];front = (front + 1) % array.length;size--;System.out.println(size); // 0 - 9notifyAll();return value;}
}

注意：
我们需要保障的是 size++ 的原子性，所以 volatile 无法解决。

import java.util.Scanner;
public class MyBlockingArrayQueueWrongDemo {// 定义个队列对象-生产者线程是 Producer，消费者线是 main 线程// 队列是需要在生产者消费者之间共享的static MyBlockingArrayQueue queue = new MyBlockingArrayQueue();// 定义一个生产者线程类static class Producer extends Thread {@Overridepublic void run() {try {int i = 0;while (true) {queue.put(i);i++;}} catch (InterruptedException e) {e.printStackTrace();}}}public static void main(String[] args) throws InterruptedException {Producer producer1 = new Producer();producer1.start();Producer producer2 = new Producer();producer2.start();Producer producer3 = new Producer();producer3.start();while (true) {queue.take();}}
}

当队列满了时，生产者线程会调用wait方法，进入warting状态，等待消费者执行完notify后对生产者线程进行唤醒。

4 管道输入/输出流

管道输入/输出流和普通的文件输入/输出流或者网络输入/输出流不同之处在于，它主要用于线程之间的数据传输，而传输的媒介为内存。

管道输入/输出流主要包括了如下4种具体实现：
面向字节

PipedOutputStream
PipedInputStream

面向字符

PipedReader
PipedWriter

printThread用来接受main线程的输入，任何main线程的输入均通过PipedWriter写入，而printThread在另一端通过PipedReader将内容读出并打印。

public class Piped {public static void main(String[] args) throws Exception {PipedWriter out = new PipedWriter();PipedReader in = new PipedReader();// 将输出流和输入流进行连接，否则在使用时会抛出IOExceptionout.connect(in);Thread printThread = new Thread(new Print(in), "PrintThread");printThread.start();int receive = 0;try {while ((receive = System.in.read()) != -1) {out.write(receive);}} finally {out.close();}}static class Print implements Runnable {private PipedReader in;public Print(PipedReader in) {this.in = in;}public void run() {int receive = 0;try {while ((receive = in.read()) != -1) {System.out.print((char) receive);}} catch (IOException ex) {}}}

注意：
对于Piped类型的流，必须先要进行绑定，也就是调用connect()方法，如果没有将输入/输出流绑定起来，对于该流的访问将会抛出异常。

5 Thread.join()

如果一个线程A执行了thread.join()语句，其含义是：当前线程A等待thread线程终止之后才从thread.join()返回。线程Thread除了提供join()方法之外，还提供了join(long millis)和join(longmillis, int nanos)两个具备超时特性的方法。这两个超时方法表示，如果线程thread在给定的超时时间里没有终止，那么将会从该超时方法中返回。

创建了10个线程，编号0～9，每个线程调用前一个线程的join()方法，也就是线程0结束了，线程1才能从join()方法中返回，而线程0需要等待main线程结束。

输出：

从上述输出可以看到，每个线程终止的前提是前驱线程的终止，每个线程等待前驱线程终止后，才从join()方法返回，这里涉及了等待/通知机制（等待前驱线程结束，接收前驱线程结束通知）。

Thread.join()方法的源码：

public final synchronized void join(long millis)throws InterruptedException {if (millis == 0) {//如果线程A还存活，那么当前线程就继续等待while (isAlive()) {wait(0);}}
}

当A线程终止时，会调用线程自身的notifyAll()方法，会通知所有等待在该线程对象上的线程。可以看到join()方法的逻辑结构与等待/通知经典范式一致，即加锁、循环和处理逻辑3个步骤。

A线程结束时调用notifyAll方法（thread.cpp）：

// 位于/hotspot/src/share/vm/runtime/thread.cpp中
void JavaThread::exit(bool destroy_vm, ExitType exit_type) {// ...// Notify waiters on thread object. This has to be done after exit() is called// on the thread (if the thread is the last thread in a daemon ThreadGroup the// group should have the destroyed bit set before waiters are notified).// 有一个贼不起眼的一行代码，就是这行ensure_join(this);// ...
}static void ensure_join(JavaThread* thread) {// We do not need to grap the Threads_lock, since we are operating on ourself.Handle threadObj(thread, thread->threadObj());assert(threadObj.not_null(), "java thread object must exist");ObjectLocker lock(threadObj, thread);// Ignore pending exception (ThreadDeath), since we are exiting anywaythread->clear_pending_exception();// Thread is exiting. So set thread_status field in  java.lang.Thread class to TERMINATED.java_lang_Thread::set_thread_status(threadObj(), java_lang_Thread::TERMINATED);// Clear the native thread instance - this makes isAlive return false and allows the join()// to complete once we've done the notify_all belowjava_lang_Thread::set_thread(threadObj(), NULL);// 唤醒其他线程lock.notify_all(thread);// Ignore pending exception (ThreadDeath), since we are exiting anywaythread->clear_pending_exception();
}

6 ThreadLocal——线程变量

public static修饰的变量为所有线程所共享，那么实现每一个线程都有自己的共享变量该如何解决——ThreadLocal（线程变量）相当于是给每个线程提供了一个局部变量。

ThreadLocal，即线程变量，是一个以ThreadLocal对象为键、任意对象为值的存储结构。这个结构被附带在线程上，也就是说每一个线程可以根据一个ThreadLocal对象查询到绑定在这个线程上的一个值。

ThreadLocal提供了线程内存储变量的能力，这些变量不同之处在于每一个线程读取的变量是对应的互相独立的。通过get和set方法就可以得到当前线程对应的值。

实际上是ThreadLocal的静态内部类ThreadLocalMap为每个Thread都维护了一个数组table，ThreadLocal确定了一个数组下标，而这个下标就是value存储的对应位置。

ThreadLocal实现主要涉及Thread，ThreadLocal，ThreadLocalMap这三个类，作为一个存储数据的类，关键点就在get和set方法。

//set 方法
public void set(T value) {//获取当前线程Thread t = Thread.currentThread();//实际存储的数据结构类型ThreadLocalMap map = getMap(t);//如果存在map就直接set，没有则创建map并setif (map != null)map.set(this, value);elsecreateMap(t, value);}//getMap方法
ThreadLocalMap getMap(Thread t) {//thred中维护了一个ThreadLocalMapreturn t.threadLocals;}//createMap
void createMap(Thread t, T firstValue) {//实例化一个新的ThreadLocalMap，并赋值给线程的成员变量threadLocalst.threadLocals = new ThreadLocalMap(this, firstValue);
}

从上面代码可以看出每个线程持有一个ThreadLocalMap对象。每一个新的线程Thread都会实例化一个ThreadLocalMap并赋值给成员变量threadLocals，使用时若已经存在threadLocals则直接使用已经存在的对象。

注意：

对于某一ThreadLocal来讲，他的索引值i是确定的，在不同线程之间访问时访问的是不同的table数组的同一位置即都为table[i]，只不过这个不同线程之间的table是独立的。
对于同一线程的不同ThreadLocal来讲，这些ThreadLocal实例共享一个table数组，然后每个ThreadLocal实例在table中的索引i是不同的。

//在某一线程声明了ABC三种类型的ThreadLocal
ThreadLocal<A> sThreadLocalA = new ThreadLocal<A>();
ThreadLocal<B> sThreadLocalB = new ThreadLocal<B>();
ThreadLocal<C> sThreadLocalC = new ThreadLocal<C>();

对于一个Thread来说只持有一个ThreadLocalMap，所以ABC对应同一个ThreadLocalMap对象。为了管理ABC，于是将他们存储在一个数组的不同位置，而这个数组就是上面提到的数组table。

ThreadLocal和Synchronized都是为了解决多线程中相同变量的访问冲突问题，不同的点是

Synchronized是通过线程等待，牺牲时间来解决访问冲突。
ThreadLocal是通过每个线程单独一份存储空间，牺牲空间来解决冲突，并且相比于Synchronized，ThreadLocal具有线程隔离的效果，只有在线程内才能获取到对应的值，线程外则不能访问到想要的值。

正因为ThreadLocal的线程隔离特性，使他的应用场景相对来说更为特殊一些。在android中Looper、ActivityThread以及AMS中都用到了ThreadLocal。当某些数据是以线程为作用域并且不同线程具有不同的数据副本的时候，就可以考虑采用ThreadLocal。

可以通过set(T)方法来设置一个值，在当前线程下再通过get()方法获取到原先设置的值。
构建一个常用的Profiler类，它具有begin()和end()两个方法，而end()方法返回从begin()方法调用开始到end()方法被调用时的时间差，单位是毫秒。

public class Profiler {// 第一次get()方法调用时会进行初始化（如果set方法没有调用），每个线程会调用一次private static final ThreadLocal<Long> TIME_THREADLOCAL = new ThreadLocal<Long>() {protected Long initialValue() {return System.currentTimeMillis();}};public static final void begin() {TIME_THREADLOCAL.set(System.currentTimeMillis());}public static final long end() {return System.currentTimeMillis() - TIME_THREADLOCAL.get();}public static void main(String[] args) throws Exception {Profiler.begin();TimeUnit.SECONDS.sleep(1);System.out.println("Cost: " + Profiler.end() + " mills");}
}结果： Cost: 1001 mills

Profiler可以被复用在方法调用耗时统计的功能上，在方法的入口前执行begin()方法，在方法调用后执行end()方法，好处是两个方法的调用不用在一个方法或者类中，比如在AOP（面向方面编程）中，可以在方法调用前的切入点执行begin()方法，而在方法调用后的切入点执行end()方法，这样依旧可以获得方法的执行耗时。

参考大佬博客：ThreadLocal