Java并发编程线程间协作（下）

Posted 2020-11-20 victorwux

上篇我们讲了使用wait()和notify()使线程间实现合作，这种方式很直接也很灵活，但是使用之前需要获取对象的锁，notify()调用的次数如果小于等待线程的数量就会导致有的线程会一直等待下去。这篇我们讲多线程间接协作的方式，阻塞队列和管道通讯，间接协作的优点是使用起来更简单并且不易出错。

阻塞队列

阻塞队列提供了一种功能，即你可以在任何时刻向队列内扔一个对象，如果队列满了则当前线程阻塞；在任何时刻都可以从队列中取出一个对象，如果队列为空则当前线程阻塞。阻塞队列是线程安全的，使用它时无需加锁。此外其内部是使用显示锁实现的同步，使用Condition实现的线程阻塞。阻塞队列的接口是BlockingQueue,它有两个实现类：

1. ArrayBlockingQueue：底层使用数组实现的队列，有固定长度，调用其构造方法时必须提供队列的最大长度。

2. LinkedBlockingQueue：底层使用链表实现的队列，理论上讲是没有最大长度的，使用时不用提供队列长度；但实际上这个队列的长度不能超过Integer.MAX_VALUE。

这两个类使用的时候没有太大区别，我们以LinkedBlockingQueue为例，重写“学生去食堂打饭”的例子，代码如下：

class Student implements Runnable {
    private Object wan = new Object();
    public void run() {
        try {
            System.out.println("学生：取到了一个碗");
            BlockingQueueTest.wanQueue.put(wan);
            System.out.println("学生：阿姨帮忙盛饭");
            wan = BlockingQueueTest.wanWithFanQueue.take();
            System.out.println("学生：吃饭");
        } catch (InterruptedException e) {}
    }
}
class CafeteriaWorker implements Runnable {
    public void run() {
        try {
            Object wan = BlockingQueueTest.wanQueue.take();
            System.out.println("阿姨：给学生盛饭");
            BlockingQueueTest.wanWithFanQueue.put(wan);
        } catch (InterruptedException e) {}
    }
}
public class BlockingQueueTest {
    public static BlockingQueue wanQueue = new LinkedBlockingQueue();
    public static BlockingQueue wanWithFanQueue = new LinkedBlockingQueue();
    public static void main(String[] args) {
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new Student());
        exec.execute(new CafeteriaWorker());
        exec.shutdown();
    }
}

输出结果如下：

学生：取到了一个碗

学生：阿姨帮忙盛饭

阿姨：给学生盛饭

学生：吃饭

在这个例子中我们定义了两个队列，一个是空碗的队列，另一个是盛完饭的碗的队列。“学生线程”取到碗后将空碗放入wanQueue队列，然后试图从wanWithFanQueue队列中取出盛好的饭碗；“阿姨线程”试图从wanQueue队列中取出空碗，然后将盛好的饭碗放到wanWithFanQueue队列中。上次我们使用wait()方法时必须要求“阿姨线程”先启动，否则会导致“阿姨线程”错过学生的信号，而使用阻塞队列实现时我们就不再要求两个线程的启动顺序了，使用阻塞队列规避了错失信号的风险。有的同学可能会好奇为什么会使用两个队列，这是因为如果使用同一个队列，同学线程把碗扔进队列后，可能“阿姨线程”没来得及取出来就被“同学线程”拿回去了，感兴趣的同学可以自行测试。

管道通讯

通过管道的方式也可以使线程间实现交互，管道和阻塞队列类似，当管道内没有数据的时候，如果某个线程尝试去读取数据就会被阻塞。

我们可以使用PipedWriter和PipedReader来实现对管道数据的读取和写入。和阻塞队列不同的是，阻塞队列中不同线程都是操作一个队列的对象；使用管道时，不同的线程可以使用不同的对象，只要将它们注册为一个管道即可。

我们使用管道通信模拟一个线程对另一个线程表白，代码如下：

class Sender implements Runnable {
    private PipedWriter writer;
    Sender(PipedWriter writer) {
        this.writer = writer;
    }
    public void run() {
        String str1 = new String("I love you
");
        String str2 = new String("Do you love me
");
        try {
            writer.write(str1.toCharArray());
            writer.write(str2.toCharArray());
        } catch (IOException e) {}
    }
}
class Receiver implements Runnable {
    private PipedReader reader;
    public Receiver(PipedReader reader) {
        this.reader = reader;
    }
    public void run() {
        try {
            while(true) {
                char c = (char)reader.read();
                System.out.print(c);
            }
        } catch (IOException e) {}
    }
}
public class PipeCommunication {
    public static void main(String[] args) throws Exception {
        PipedReader reader = new PipedReader();
        PipedWriter writer = new PipedWriter(reader);
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new Sender(writer));
        exec.execute(new Receiver(reader));
                Thread.sleep(1000);
        exec.shutdownNow();
    }
}

运行后输出结果如下，一秒后程序退出：

I love you

Do you love me

我们在主方法里先定义了一个PipedReader对象，然后将这个对象作为PipedWriter的构造方法的参数传给PipedWriter对象，这样就实现两个输入输出流的绑定，分别将两个流对象传给两个线程对象。在信息的接收方我们使用一个死循环让其不断的从管道内读入，从输出结果可以看出read()方法在管道内没有数据的时候被阻塞了，因为输出结果没有循环打印其它字符。此外主线程sleep一秒后调用了shutdownNow()方法，这个方法向所有运行着的线程发送中断信号，程序运行一秒后就退出了，我们可以看出中断信号打断了Receiver的阻塞状态，由此得出结论：管道类阻塞时可以被中断信号打断。

总结

本篇讲了使用阻塞队列和管道来实现线程间的合作，相对于使用wait()协作而言这两种方式更为高级，使用起来更容易而且不易错，此外阻塞队列和管道都是线程安全的，因此使用它们的时候不需要使用锁。需要实现线程间协作时可以根据实际需要，权衡利弊进行选择。

公众号：今日说码。关注我的公众号，可查看连载文章。遇到不理解的问题，直接在公众号留言即可。