16.线程通信1：生产者/消费者问题

Posted 2022-09-06 纵横千里，捭阖四方

线程之间的通信又称为线程同步，是指当某个线程修改了一个对象的值时，另外一个线程能感知到该值的变化并进行相应的操作。

实现线程之间通信的方法有：

1.基于volatile修饰的共享变量

2.通过wait和notify机制

3.Thread.Join方法

4.使用synchronized同步关键字

5.Condition.await/signal方法

Java提供了（wait/notify）等待/通知机制来实现多个线程之间的协同处理，也就是控制线程之间的等待和唤醒。

wait()方法，使当前线程进入阻塞状态，并且释放持有的锁。

notify()方法，唤醒处于阻塞状态的下一个线程

notifyAll()方法，唤醒所有处于阻塞状态下的线程。

线程的通信（wait/notify）等待/通知机制，是指一个线程A调用了对象O的wait()方法进入等待状态，而另一个线程B调用了对象O的notify()或者notifyAll()方法，线程A收到通知后从对象O的wait()方法返回，进而执行后续操作。上述两个线程通过对象O来完成交互，而对象上的wait()和notify/notifyAll()的关系就如同开关信号一样，用来完成等待方和通知方之间的交互工作。

1 生产者/消费者问题

生产者/消费者是经典的线程通信与协同机制，接下来，我们设计一个简单的模型。

生产者：

public class Product implements Runnable 
    private Queue<String> msg;
    private int maxSize;

    public Product(Queue<String> msg, int maxSize) 
        this.msg = msg;
        this.maxSize = maxSize;
    
    @Override
    public void run() 
        int i=0;
        while (true) 
            synchronized (msg) 
                while (msg.size() == maxSize) 
                    //生产者满了
                    try 
                        msg.wait();
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                
                try 
                    Thread.sleep(1000);
                 catch (InterruptedException e) 
                    e.printStackTrace();
                

                System.out.println("生产...");
                msg.add("生产消息"+i);
                msg.notify();
                msg.notifyAll();
            
        
    

Producer表示一个生产者线程，该线程定义了一个共享对象msg，然后在run()方法中实现如下逻辑：

• 使用synchronized(msg)对共享对象加锁。

• 当msg.size==maxSize时，表示队列满了，让当前线程等待。

• 否则往bags中添加数据，并且使用msg.notify()方法唤醒阻塞的消费者线程。

从代码中可以发现，wait()和notify()方法必须要写在synchronized代码块中，至于原因 ，后面再分析。

消费者：

public class Consumer implements Runnable 
    private Queue<String>msg;
    private int maxSize;

    public Consumer(Queue<String> msg, int maxSize) 
        this.msg = msg;
        this.maxSize = maxSize;
    

    @Override
    public void run() 
        while (true)
            synchronized (msg)
                //消费者空了
                while (msg.isEmpty())
                    try 
                        msg.wait();//阻塞当前队列
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                

            try 
                Thread.sleep(1000);
             catch (InterruptedException e) 
                e.printStackTrace();
            
            System.out.println("消费者消费消息"+msg.remove());
            msg.notify();//唤醒处于等待状态的生产者
            
        
    

Consumer表示一个消费者线程，同样会用到共享对象msg，接着在run()方法中实现如下逻辑：

• 先使用synchronized(msg)方法加锁，注意生产者/消费者锁定的对象实例必须是同一个。

• 如果msg.isEmpty()表示队列空了，则需要等待生产者生产才能继续消费。

• 否则，使用msg.remove()方法从队列中消费数据，当数据消费了之后，可以使用msg.notify()方法唤醒处于阻塞状态的生产者线程。

写个测试代码，main方法：

public class TestMain 
    public static void main(String[] args) 
        Queue<String>queue=new LinkedList<>();
        int maxSize=2;
        Product product=new Product(queue,maxSize);
        Consumer consumer=new Consumer(queue,maxSize);

        Thread t1=new Thread(product);
        Thread t2=new Thread(consumer);
        t1.start();
        t2.start();
    

生产者先生产数据，然后才能唤醒消费者线程，在生产了两个数据之后发现队列满了，此时通过wait()方法阻塞生产者线程。消费者线程开始运行，如果发现队列中的元素不为空，则进行数据的消费，同时唤醒生产者线程，如果消费者发现队列为空 ，则阻塞消费者线程。

生产者/消费者如果深入设计，可以非常复杂，例如，假如有多个消费者或者生产者该如何设计等等。

2 图解生产者/消费者

上面的生产者/消费者模型的执行过程是怎么样的呢？

我们这里用queue做为两者的共享队列，生产者和消费者都会对该共享队列进行读写操作。因此，为了保证原子性，生产者和消费者线程就必须要对该共享队列加锁，只有竞争到锁资源的线程才能加锁，也就是操作队列的数据。

假设生产者抢到了锁，则开始向队列添加数据，知道队列满了发生阻塞，也就是消费应该要消费数据了。在代码中生产者的wait()方法是在synchronized中调用的，此时锁是没有释放的，那么消费者获取不到同步锁该怎么办呢？其实在线程调用了wait()方法后，就会释放当前的同步锁。由于Consumer此时在synchronized同步队列中等待，所以Produer一旦释放锁就可以唤醒Consumer线程。Consumer被唤醒后需要去竞争锁资源，如果成功，则进行数据消费。

Consumer消费一个数据后会调用notify()方法，该方法只是唤醒处于阻塞状态下的线程，由于Consumer还没有释放锁，因此被唤醒的Producer需要等待Consumer释放锁之后才能继续沿着阻塞的位置开始执行。

过程图如下：