线程间通信-白红宇

线程间通信

阅读量：4665 次

发布时间：2019-06-09

本文共 14127 字，大约阅读时间需要 47 分钟。

　　在我们使用多个线程来同时运行多个任务时，可以通过使用锁(互斥)来同步两个或多个任务的行为，从而使得一个任务不会干涉另一个任务使用的资源。

　　这个问题已经解决了，下一步是学习如何使任务彼此之间可以协作，以使得多个任务可以一起工作去解决某个问题。在这类问题中不可避免会碰到某些部分必须在其他部分被解决之前解决。在解决这类问题时，关键是多个任务之间如何“握手”，即通过何种方式来通知对方。在Java中有多种方式以及工具可以帮助我们实现这种“握手”。方式比较多，总结了一下，主要如下：

1. 中断和检查中断机制

　　利用中断和加入线程(join)属于Thread提供的原生机制，用来实现线程之间的通信。

　　关于中断，Java中的线程是通过中断状态来标识，可以通过线程的实例方法interrupt()来设置(该方法内部会调用一个native方法interrupt0()进行设置)。中断状态只是一个状态，并不会直接影响线程的运行，Java中是通过一些间接的方式来达到控制线程的效果，比如：

检查在循环中断状态，判断是否需要退出线程；

或者通过对中断状态进行监控，用一旦发现改变就抛出异常这种方式来通知程序；

　　我们通过两个实例看一下：

public class ThreadDemo {    public static void main(String[] args){                Thread t = new Thread(new ThreadCheckInterrupt());        t.start();        try {            Thread.sleep(100);        }catch (InterruptedException e) {                }        t.interrupt();            }        static class ThreadCheckInterrupt implements Runnable{                @Override        public void run() {            System.out.println("thread start --" + System.currentTimeMillis());            while(!Thread.interrupted()) {            }            System.out.println("thread interrupt -- " + System.currentTimeMillis());        }            }}/** * 输出结果 **/thread start --1556775343297thread interrupt -- 1556775343407

　　上面的例子中，主线程启动子线程t，子线程进入循环，判断条件是!Thread.interrupted()，每次循环开始都检查一下中断状态，如果中断状态被设置了，则退出循环，这就是检查中断状态机制，这里例子中是在主线程中调用子线程的interrupt()方法来设置子线程中断状态的。

public class ThreadDemo {    public static void main(String[] args){        Thread t = new Thread(new ThreadInterruptDemo());        t.start();        try {            Thread.sleep(100);        }catch (InterruptedException e) {                }        t.interrupt();            }        static class ThreadInterruptDemo implements Runnable{        @Override        public void run() {            try {                Thread.sleep(1000);            }catch (InterruptedException e) {                System.out.println("interrupt works");            }        }           }}/** * 输出结果 **/ interrupt works

　　在这个例子中，主线程仍然开启一个子线程，主线程休眠100ms，子线程休眠1000ms，确保主线程能够在子线程休眠时设置其中断状态，以触发中断异常的抛出。这种通过中断来触发异常来通知程序的方式即是Java中的中断机制。

　　在调用wait()、sleep()、join()等方法导致的线程阻塞期间，如果有其他线程调用了该线程的interrupt()来中断该线程，则会导致前面这种阻塞状态停止，抛出InterruptedException异常，并且其中断状态会复原。

2. 加入线程(Join)

　　join()方法也是Thread提供的一个实例方法，提供了一个类似线程之间等待的效果。一个线程可以调用另一个线程的join()方法，调用之后，该线程将进入等待状态，直到另一个线程执行完毕，该线程才会继续执行，示例如下：

public class ThreadDemo {    public static void main(String[] args){        System.out.println("main thread start , time -- > " + System.currentTimeMillis());        Thread t = new Thread(new ThreadJoin());        t.start();        try {            t.join();        } catch (InterruptedException e) {}        System.out.println("main thread end , time -- > " + System.currentTimeMillis());    }        static class ThreadJoin implements Runnable{        @Override        public void run() {            System.out.println("sub thread start");            try {                Thread.sleep(1000);            }catch (InterruptedException e) {                            }            System.out.println("sub thread end");        }    }}/** * 输出结果 **/main thread start , time -- > 1556801790138sub thread startsub thread endmain thread end , time -- > 1556801791138

　　主线程开始之后，启动子线程t，并调用t的join()方法，可以看到，直到子线程执行结束，主线程才继续执行，这就实现了线程之间的等待效果。

　　join()方法也有带参数的重载版本，可以指定等待的时间，即使当超过指定时间等待的线程仍然没有执行结束，join()方法将返回，不会继续等待。

3. 互斥+信号(synchronized、ReentrantLock)

　　Java中线程之间的互斥主要是借助于synchronized、ReentrantLock来实现，互斥既可以保证任务之间的串行，也可以保证某一时刻只有一个任务可以响应某个信号。在互斥之上，我们为任务添加了一种途径，可以将其自身挂起，直至某些外部条件发生变化，表示是时候让这个任务向前继续执行。这种握手可以通过Object的方法wait()和notify()来安全地实现，也可以通过Java SE5中并发类库提供的具有await()和signal()方法的Condition对象。

　　通过synchronized实现的互斥和Object提供的wait()、notify()/notifyAll()方法互相配合，可以实现线程之间的等待通知机制。调用wait()、notify()等方法的对象必须为对象锁，且必须在同步块内执行调用。调用wait()方法的线程将进入等待状态，且会释放已经获取的同步锁；调用notify()/notifyAll()方法的线程将会通知处于等待状态的线程，从等待状态切换到准备竞争同步锁，一旦抢到锁则继续从之前等待的地方(即wait()处)执行：

public class ThreadDemo{    static String content;    static String LOCK = "lock";    public static void main(String[] args){        new Thread(){            @Override            public void run(){                synchronized(LOCK){                    content = "hello world";                    LOCK.notifyAll();                }            }         }.start();        synchronized(LOCK){            while(content == null){                try{                    LOCK.wait();                }catch(InterruptedException e){                    e.printStackTrace();                }                System.out.println(content.toUpperCase());            }        }    }}// 输出HELLO WORLD

　　而ReentrantLock提供的互斥是和其内部类Condition提供的await()、signal()相配合来实现线程之间的等待通知机制。与上面一样，await()、signal()方法必须在获取锁的情况下才能够调用，否则会抛出异常。Condition实例对象通过ReentrantLock对象获取。

public class ReentrantLockCondition {        public static final ReentrantLock lock = new ReentrantLock();        public static final Condition c = lock.newCondition();        public static void main(String[] args) {        System.out.println("main thread start , time -- > " + System.currentTimeMillis());        Thread sub = new Thread(new ConditionTest());        try {            lock.lock(); 　　　　　　　sub.start();            c.await();        } catch (InterruptedException e) {        }finally {            lock.unlock();        }        System.out.println("main thread end , time -- > " + System.currentTimeMillis());    }        static class ConditionTest implements Runnable{        @Override        public void run() {            System.out.println("sub thread start , time -- > " + System.currentTimeMillis());            try {                lock.lock();                Thread.sleep(1000);                c.signal();            }catch(InterruptedException e) {                            }finally {                lock.unlock();            }            System.out.println("sub thread end , time -- > " + System.currentTimeMillis());        }    }}/** * 输出结果 **/main thread start , time -- > 1556883481141sub thread start , time -- > 1556883481141sub thread end , time -- > 1556883482141main thread end , time -- > 1556883482141

　　如上，主线程获取锁，启动子线程，并调用await()方法进入阻塞状态，并释放锁，这时子线程会获取锁，休眠1s之后调用signal()方法通知主线程，子线程释放锁之后，主线程可以获取锁，获取成功之后继续执行。所以结果是先有主线程启动的日志，然后是子线程启动和结束的日志，最后则是主线程结束的日志。

4. 利用并发工具包中的构件(CountLatchDown、CyclicBarrier)

　　java.util.concurrent包在JDK1.5之后引入了大量用来解决并发问题的新类，比如CountDownLatch、CyclicBarrier等，各有自己的特点，我们仍然看一些例子：

4.1 CountLatchDown

　　它可以用来同步一个或多个任务，强制它们等待由其他任务执行的一组操作完成，用法入下：

public class CountDownLatchSimpleDemo {        public static CountDownLatch c = new CountDownLatch(2);        public static void main(String[] args) {        System.out.println("main thread start , time -> " + System.currentTimeMillis());        Thread t1 = new Thread(new Count());        Thread t2 = new Thread(new Count());        t1.start();        t2.start();        try {            c.await();        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("main thread end , time -> " + System.currentTimeMillis());    }        static class Count implements Runnable{                private static int counter = 0;        private final int id = counter++;                @Override        public void run() {            System.out.println("thread " + id + " start , time -> " + System.currentTimeMillis());            try {                Thread.sleep(1000);            }catch(InterruptedException e) {                            }            System.out.println("thread " + id + " end , time -> " + System.currentTimeMillis());            c.countDown();        }    }}/** * 输出结果 **/main thread start , time -> 1557065917966thread 0 start , time -> 1557065917968thread 1 start , time -> 1557065917968　　thread 0 end , time -> 1557065918968thread 1 end , time -> 1557065918968main thread end , time -> 1557065918968

　　主线程启动两个子线程，然后调用await()方法之后进入等待状态，两个子线程启动之后会调用CountDownLatch的countDown()方法，总共调用了两次，就将CountDownLatch对象初始化时指定的计数2消去，这时候主线程就会从等待状态中被唤醒，继续执行。

4.2 CyclicBarrier

　　这个东西江湖人称栅栏，可以协调多个线程一起运行，实现类似多个线程都达到某个状态之后再继续运行的效果。实现机制简单介绍下：

首先将要参与的线程初始化；

调用await()方法的线程将会进入等待状态；

直到指定数量的线程都调用了await()方法，之前处于等待状态的线程才会从调用await()处继续执行；

public class CyclicBarrierSimpleDemo {        public static AtomicInteger id = new AtomicInteger();        public static AtomicInteger count = new AtomicInteger();        public static CyclicBarrier barrier = new CyclicBarrier(2, ()->{        if(count.get() > 2) {            return;        }        count.incrementAndGet();        System.out.println("barrier trip " + count.get());        }    );        public static void main(String[] args) {        Thread t1 = new Thread(new SubTask());        Thread t2 = new Thread(new SubTask());        t1.start();        t2.start();        try {            Thread.sleep(500);        }catch(InterruptedException e) {                    }        t1.interrupt();        t2.interrupt();    }        static class SubTask implements Runnable{                int threadId = id.getAndIncrement();                @Override        public void run() {            while(!Thread.interrupted()) {                try {                    Thread.sleep(100);                    System.out.println("thread " + threadId + " play");                    barrier.await();                } catch (InterruptedException e) {                    System.out.println("thread " + threadId + " interrupt");                    return;                } catch (BrokenBarrierException e) {                    e.printStackTrace();                }            }            System.out.println("thread " + threadId + " end");        }    }}/**  * 运行结果  **/thread 0 playthread 1 playbarrier trip 1thread 1 playthread 0 playbarrier trip 2thread 1 playthread 0 playbarrier trip 3thread 1 playthread 0 playthread 0 interruptthread 1 interrupt

　　如上是一个简单的demo，主线程会启动两个子线程，子线程会调用CyclicBarrier的await()方法，从上面输出结果可以看出，子线程之间会互相等待，执行一次，越过一次栅栏。

5. 共享内存变量(volatile)

　　volatile关键字可以保证可见性和有序性，这里也是利用其可见性来实现线程之间的通信的，因为被volatile修饰的变量一旦发生变化，对其它线程是可见的。通过监控某个共享变量，当其状态发生改变时，就可以认为是收到别的线程的信号了。

public class VolatileDemo {    public static volatile boolean index = false;        public static void main(String[] args) {        Thread sub = new Thread(new SubThread());        sub.start();        System.out.println("main thread notify sub thread , time -- > " + System.currentTimeMillis());        try {            Thread.sleep(200);        }catch(InterruptedException e) {                    }        index = true;    }}class SubThread implements Runnable{    @Override    public void run() {        System.out.println("sub thread start , time -- > " + System.currentTimeMillis());        while(!VolatileDemo.index) {        }        System.out.println("sub thread end , time -- > " + System.currentTimeMillis());    }}/** * 运行结果 **/main thread notify sub thread , time -- > 1557145592276sub thread start , time -- > 1557145592276sub thread end , time -- > 1557145592477

　　如上面的例子，线程一直监控共享变量index，当主线程将index改为true时，子线程可以马上监控到，就可以退出while循环了。

6. 管道(PipedWriter、PipedReader)

　　通过IO阻塞的方式，我们也能实现线程之间的通信：

public class PipeNotifyDemo {        public static void main(String[] args) throws IOException {        PipedWriter writer = new PipedWriter();        PipedReader reader = new PipedReader();        writer.connect(reader);                Thread t1 = new Thread(()->{            System.out.println("writer running");            try {                for(int i = 0 ; i < 5 ; i++) {                    writer.write(i);                    Thread.sleep(1000);                }            } catch (Exception e) {                e.printStackTrace();            }finally {                try {                    writer.close();                } catch (IOException e) {                    e.printStackTrace();                }            }            System.out.println("writer ending");        });                Thread t2 = new Thread(()->{            System.out.println("reader running");            int message = 0;            try {                while((message = reader.read()) != -1) {                    System.out.println("message = " + message + " , time -- > " + System.currentTimeMillis());                }            }catch(Exception e) {            }finally {                try {                    reader.close();                } catch (IOException e) {                    e.printStackTrace();                }            }            System.out.println("reader ending");        });                t1.start();        t2.start();    }}/** * 输出结果 **/reader runningwriter runningmessage = 0 , time -- > 1557148872499message = 1 , time -- > 1557148872499message = 2 , time -- > 1557148873499message = 3 , time -- > 1557148875499message = 4 , time -- > 1557148875499writer endingreader ending

　　线程2源源不断收到线程1写入的数据。

7. 通过线程池的一个方法(awaitTermination方法)

　　线程池中有一个方法awaitTermination()方法，可以用来判断调用线程池的shutdown()方法之后，线程池中的线程是否执行完毕。因为shutdown()方法是不会阻塞的，调用之后线程池不会接受新任务，但是已有的任务会继续执行，所以通过awaitTermination方法来判断是否存在还在执行的任务，这也算是一种线程之前的通信吧。

public class ThreadPoolNotifyDemo {        public static void main(String[] args) throws InterruptedException {        ExecutorService exec = Executors.newCachedThreadPool();        exec.execute(()->{            System.out.println("thread1 running");            try {                Thread.sleep(2000);            }catch(InterruptedException e) {                e.printStackTrace();            }        });                exec.execute(()->{            System.out.println("thread2 running");            try {                Thread.sleep(3000);            }catch(InterruptedException e) {                e.printStackTrace();            }        });                exec.shutdown();        while(!exec.awaitTermination(1, TimeUnit.SECONDS)) {            System.out.println("thread in thread pool is still running , time -- > " + System.currentTimeMillis());        }        System.out.println("main thread over");    }}/** * 输出结果 **/thread1 runningthread2 runningthread in thread pool is still running , time -- > 1557150326140thread in thread pool is still running , time -- > 1557150327141main thread over