Java多线程系列—生产者消费者模式的多种实现(04)

迷南。 2022-10-21 14:58 125阅读 0赞

## 生产者消费者模式 ##

我们先来看看什么是生产者消费者模式，生产者消费者模式是程序设计中非常常见的一种设计模式，**被广泛运用在解耦、消息队列等场景**。

在现实世界中，我们把生产商品的一方称为生产者，把消费商品的一方称为消费者，有时生产者的生产速度特别快，但消费者的消费速度跟不上，俗称“产能过剩”，又或是多个生产者对应多个消费者时，大家可能会手忙脚乱。如何才能让大家更好地配合呢？这时在生产者和消费者之间就需要一个中介来进行调度，于是便诞生了生产者消费者模式。

![img][]

使用生产者消费者模式通常需要在两者之间增加一个阻塞队列作为媒介，有了媒介之后就相当于有了一个缓冲，平衡了两者的能力，整体的设计如图所示，最上面是阻塞队列，右侧的 1 是生产者线程，生产者在生产数据后将数据存放在阻塞队列中，左侧的 2 是消费者线程，消费者获取阻塞队列中的数据。而中间的 3 和 4 分别代表生产者消费者之间互相通信的过程，因为无论阻塞队列是满还是空都可能会产生阻塞，阻塞之后就需要在合适的时机去唤醒被阻塞的线程。

那么什么时候阻塞线程需要被唤醒呢？有两种情况。第一种情况是当消费者看到阻塞队列为空时，开始进入等待，这时生产者一旦往队列中放入数据，就会通知所有的消费者，唤醒阻塞的消费者线程。另一种情况是如果生产者发现队列已经满了，也会被阻塞，而一旦消费者获取数据之后就相当于队列空了一个位置，这时消费者就会通知所有正在阻塞的生产者进行生产，这便是对生产者消费者模式的简单介绍。

下面我们看一下如何实现这个模式

### BlockingQueue 实现生产者消费者 ###

我们接下来看如何用 wait/notify/Condition/BlockingQueue 实现生产者消费者模式，先从最简单的 BlockingQueue 开始讲起：

public class ProducerConsumerPatternSimple { 
        public static void main(String[] args) { 
            BlockingQueue<Object> queue = new ArrayBlockingQueue<>(10);
            Runnable producer = () -> { 
                while (true) { 
                    try { 
                        Object object=new Object();
                        queue.put(object);
                        System.out.println("生产了:"+object.toString());
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
    
            new Thread(producer).start();
            new Thread(producer).start();
    
            Runnable consumer = () -> { 
                while (true) { 
                    try { 
                        Object object = queue.take();
                        System.out.println("消费了:"+object.toString());
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
            new Thread(consumer).start();
            new Thread(consumer).start();
        }
    
    }

如代码所示，首先，创建了一个 ArrayBlockingQueue 类型的 BlockingQueue，命名为 queue 并将它的容量设置为 10；其次，创建一个简单的生产者，while(true) 循环体中的queue.put() 负责往队列添加数据；然后，创建两个生产者线程并启动；同样消费者也非常简单，while(true) 循环体中的 queue.take() 负责消费数据，同时创建两个消费者线程并启动。以上便是利用 BlockingQueue 实现生产者消费者模式的代码。

虽然代码非常简单，但实际上 ArrayBlockingQueue 已经在背后完成了很多工作，**比如队列满了就去阻塞生产者线程，队列有空就去唤醒生产者线程等**。

### Condition 实现生产者消费者模式 ###

BlockingQueue 实现生产者消费者模式看似简单，背后却暗藏玄机，我们在掌握这种方法的基础上仍需要掌握更复杂的实现方法。我们接下来看如何在掌握了 BlockingQueue 的基础上利用 Condition 实现生产者消费者模式，它们背后的实现原理非常相似，相当于我们自己实现一个简易版的 BlockingQueue：

public class BlockingQueueForCondition { 
        private Queue queue;
        private int max = 16;
        private ReentrantLock lock = new ReentrantLock();
        private Condition notEmpty = lock.newCondition();
        private Condition notFull = lock.newCondition();
    
        public static void main(String[] args) { 
            BlockingQueueForCondition blockingQueueForCondition = new BlockingQueueForCondition(16);
            new Thread(() -> { 
                while (true) { 
                    Object o = new Object();
                    System.out.println("生产了:" + o.toString());
                    try { 
                        blockingQueueForCondition.put(o);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }).start();
    
    
            new Thread(() -> { 
                while (true) { 
                    try { 
                        Object o = blockingQueueForCondition.take();
                        System.out.println("消费了:" + o.toString());
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }).start();
    
        }
    
    
        public BlockingQueueForCondition(int size) { 
            this.max = size;
            queue = new LinkedList();
        }
    
        public void put(Object o) throws InterruptedException { 
            lock.lock();
            try { 
                while (queue.size() == max) { 
                    notFull.await();
                }
                queue.add(o);
                notEmpty.signalAll();
            } finally { 
                lock.unlock();
            }
        }
    
        public Object take() throws InterruptedException { 
            lock.lock();
            try { 
                while (queue.size() == 0) { 
                    notEmpty.await();
                }
                Object item = queue.remove();
                notFull.signalAll();
                return item;
            } finally { 
                lock.unlock();
            }
        }
    }

如代码所示，首先，定义了一个队列变量 queue 并设置最大容量为 16,需要注意的是我们并没有使用阻塞队列，而是使用了普通队列；其次，定义了一个 ReentrantLock 类型的 Lock 锁，并在 Lock 锁的基础上创建两个 Condition，一个是 notEmpty，另一个是 notFull，分别代表队列没有空和没有满的条件；最后，声明了 put 和 take 这两个核心方法。

**因为生产者消费者模式通常是面对多线程的场景，需要一定的同步措施保障线程安全，所以在 put 方法中先将 Lock 锁上**，然后，在 while 的条件里检测 queue 是不是已经满了，如果已经满了，则调用 notFull 的 await() 阻塞生产者线程并释放 Lock，如果没有满，则往队列放入数据并利用 notEmpty.signalAll() 通知正在等待的所有消费者并唤醒它们。最后在 finally 中利用 lock.unlock() 方法解锁，把 unlock 方法放在 finally 中是一个基本原则，否则可能会产生无法释放锁的情况。

下面再来看 take 方法，take 方法实际上是与 put 方法相互对应的，同样是通过 while 检查队列是否为空，如果为空，消费者开始等待，如果不为空则从队列中获取数据并通知生产者队列有空余位置，最后在 finally 中解锁。

这里需要注意，我们在 take() 方法中使用 while( queue.size() == 0 ) 检查队列状态，而不能用 if( queue.size() == 0 )。为什么呢？大家思考这样一种情况，因为生产者消费者往往是多线程的，我们假设有两个消费者，第一个消费者线程获取数据时，发现队列为空，便进入等待状态；因为第一个线程在等待时会释放 Lock 锁，所以第二个消费者可以进入并执行 if( queue.size() == 0 )，也发现队列为空，于是第二个线程也进入等待；而此时，如果生产者生产了一个数据，便会唤醒两个消费者线程，而两个线程中只有一个线程可以拿到锁，并执行 queue.remove 操作，另外一个线程因为没有拿到锁而卡在被唤醒的地方，而第一个线程执行完操作后会在 finally 中通过 unlock 解锁，而此时第二个线程便可以拿到被第一个线程释放的锁，继续执行操作，也会去调用 queue.remove 操作，然而这个时候队列已经为空了，所以会抛出 NoSuchElementException 异常，这不符合我们的逻辑。而如果用 while 做检查，当第一个消费者被唤醒得到锁并移除数据之后，第二个线程在执行 remove 前仍会进行 while 检查，发现此时依然满足 queue.size() == 0 的条件，就会继续执行 await 方法，避免了获取的数据为 null 或抛出异常的情况。

其实关于这个我们使用while 进行判断的写法我们前面已经说过了，就是为了唤醒之后是否满足运行条件的检测。使用lock 和信号量的配合的特点在于灵活，我们可以利用lock 的多个信号量完成复杂的逻辑

### wait/notify 实现生产者消费者模式 ###

最后我们再来看看使用 wait/notify 实现生产者消费者模式的方法，实际上实现原理和Condition 是非常类似的，采取的思想都是主动通知，也就是说和队列的实现逻辑对比起来，更有主动性，但是也意味着承担更多的责任。

public class BlockingQueueForWaitAndNotify { 
        private int maxSize;
        private LinkedList<Object> storage;
    
        public static void main(String[] args) { 
            BlockingQueueForWaitAndNotify blockingQueueForWaitAndNotify = new BlockingQueueForWaitAndNotify(16);
    
            new Thread(()->{ 
                while (true){ 
                    try { 
                        blockingQueueForWaitAndNotify.put();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
    
                }
            }).start();
    
    
            new Thread(()->{ 
                while (true){ 
                    try { 
                        blockingQueueForWaitAndNotify.take();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
    
                }
            }).start();
        }
    
        public BlockingQueueForWaitAndNotify(int size) { 
            this.maxSize = size;
            storage = new LinkedList<>();
        }
    
        public synchronized void put() throws InterruptedException { 
            while (storage.size() == maxSize) { 
                wait();
            }
            Object o = new Object();
            storage.addLast(o);
            System.out.println("生产了:"+o);
            notifyAll();
        }
    
        public synchronized void take() throws InterruptedException { 
            while (storage.size() == 0) { 
                wait();
            }
            System.out.println("消费了:"+storage.removeFirst());
            notifyAll();
        }
    }

如代码所示，最主要的部分仍是 take 与 put 方法，我们先来看 put 方法，put 方法被 synchronized 保护，while 检查队列是否为满，如果不满就往里放入数据并通过 notifyAll() 唤醒其他线程。同样，take 方法也被 synchronized 修饰，while 检查队列是否为空，如果不为空就获取数据并唤醒其他线程。

其实看到这里你可能已经发现了Condition和wait/notify 的区别了，其实就是wait/notify 唤醒了在等待的全部线程，包括生产线程和消费线程，而Condition通过两个信号量灵活的控制了这种场景，例如生产之后只唤醒消费线程，其实Condition的信号量还可以更多从而更加的灵活控制程序，其实你已经知道原来一个Condition也是可以玩转生产者消费者模型的，其实就是模拟wait/notify

### Condition 模拟wait/notify ###

public class BlockingQueueForConditionSimple { 
        private Queue queue;
        private int max = 16;
        private ReentrantLock lock = new ReentrantLock();
        private Condition condition = lock.newCondition();
    
        public static void main(String[] args) { 
            BlockingQueueForConditionSimple blockingQueueForCondition = new BlockingQueueForConditionSimple(10);
            new Thread(() -> { 
                while (true) { 
                    Object o = new Object();
                    System.out.println("生产了:" + o.toString());
                    try { 
                        blockingQueueForCondition.put(o);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }).start();
    
    
            new Thread(() -> { 
                while (true) { 
                    try { 
                        Object o = blockingQueueForCondition.take();
                        System.out.println("消费了:" + o.toString());
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }).start();
    
        }
    
    
        public BlockingQueueForConditionSimple(int size) { 
            this.max = size;
            queue = new LinkedList();
        }
    
        public void put(Object o) throws InterruptedException { 
            lock.lock();
            try { 
                while (queue.size() == max) { 
                    condition.await();
                }
                queue.add(o);
                condition.signalAll();
            } finally { 
                lock.unlock();
            }
        }
    
        public Object take() throws InterruptedException { 
            lock.lock();
            try { 
                while (queue.size() == 0) { 
                    condition.await();
                }
                Object item = queue.remove();
                condition.signalAll();
                return item;
            } finally { 
                lock.unlock();
            }
        }
    }

我们只使用了一个信号量的两个方法就完成了生产者消费者模式，这个设计和wait/notify的实现是一模一样的。

### Disruptor生产者消费者的终极解决方案 ###

Disruptor是一个开源的JAVA框架，它被设计用于在生产者—消费者（producer-consumer problem，简称PCP）问题上获得尽量高的吞吐量（TPS）和尽量低的延迟。Disruptor是LMAX在线交易平台的关键组成部分，LMAX平台使用该框架对订单处理速度能达到600万TPS，除金融领域之外，其他一般的应用中都可以用到Disruptor，它可以带来显著的性能提升。其实Disruptor与其说是一个框架，不如说是一种设计思路，这个设计思路对于存在“并发、缓冲区、生产者—消费者模型、事务处理”这些元素的程序来说，Disruptor提出了一种大幅提升性能（TPS）的方案。

Disruptor是一个高性能的异步处理框架，或者可以认为是最快的消息框架（轻量的JMS），也可以认为是一个观察者模式的实现，或者事件监听模式的实现。  
关于Disruptor我们后面还会有详细介绍，这里我们只介绍生产者消费者，其实从上面的实现上我们发现了一个问题，先看看下面的图，你以为你实现的模式是下面这样的

![image-20210502112802102][]

其实你实现的是这样的，也就是说你的生产或者或者是消费这个行为是多线程的，但是你的放入队列和从队列读取的这个过程依然是单线程的——你加了锁

![image-20210502113258553][]

而Disruptor做的就是优化就是通过Ring Buffer配合CAS操作避免了加锁这个操作，也就是说可以在无锁的情况下对队列进行操作，那么这个队列的设计就是Disruptor的核心所在

public class DisruptorMain { 
        public static void main(String[] args) { 
            // 1.创建一个可缓存的线程 提供线程来出发Consumer 的事件处理
            ExecutorService executor = Executors.newCachedThreadPool();
            // 2.创建工厂
            EventFactory<LongEvent> eventFactory = new LongEventFactory();
            // 3.创建ringBuffer 大小 ringBufferSize大小一定要是2的N次方
            int ringBufferSize = 1024 * 1024;
            // 4.创建Disruptor
            Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(eventFactory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
            // 5.连接消费端方法
            disruptor.handleEventsWith(new LongEventConsumer());
            //6.启动
            disruptor.start();
            // 7.创建RingBuffer容器
            RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
            // 8.创建生产者
            LongEventProducer producer = new LongEventProducer(ringBuffer);
            producer.pushData();
            disruptor.shutdown();
            executor.shutdown();
    
        }
    
    
        static class LongEventProducer { 
            public final RingBuffer<LongEvent> ringBuffer;
    
            public LongEventProducer(RingBuffer<LongEvent> ringBuffer) { 
                this.ringBuffer = ringBuffer;
            }
    
    
            public void pushData() { 
                Random random = new Random();
                while (true) { 
                    // 1.ringBuffer 事件队列 下一个槽
                    long sequence = ringBuffer.next();
                    long data = random.nextLong();
                    try { 
                        //2.取出空的事件队列
                        LongEvent longEvent = ringBuffer.get(sequence);
                        //3.获取事件队列传递的数据
                        longEvent.setValue(data);
                    } finally { 
                        System.out.println("生产者生产数据:" + data);
                        //4.发布事件
                        ringBuffer.publish(sequence);
    
                    }
                }
    
            }
        }
    
        static class LongEventConsumer implements EventHandler<LongEvent> { 
            @Override
            public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception { 
                System.out.println("消费者消费数据:" + event.getValue());
            }
        }
    
        // 让Disruptor为我们创建事件，我们同时还声明了一个EventFactory来实例化Event对象
        static class LongEventFactory implements EventFactory<LongEvent> { 
            @Override
            public LongEvent newInstance() { 
                return new LongEvent();
            }
    
        }
    
        // 首先声明一个Event来包含需要传递的数据
        static class LongEvent { 
            private Long value;
    
            public Long getValue() { 
                return value;
            }
    
            public void setValue(Long value) { 
                this.value = value;
            }
    
        }
    }

上面就是我们实现的一个简单的Disruptor版本的生产者和消费者，特点就是在整个过程中你都没有使用锁。Disruptor有很多应用，类似Log4j2

Log4j2相对于Log4j1最大的优势在于多线程并发场景下性能更优。该特性源自于Log4j 2的异步模式采用了Disruptor来处理。

## 总结 ##

1.  使用阻塞式队列的方式最简单
2.  使用Condition 的方式最灵活也最复杂
3.  使用wait/notify的方式灵活性没有Condition那么高，同样复杂性也没那么高。需要明白condition 可以实现wait/notify的效果，但是反过来就不行
4.  生产者消费者是一种设计模式，主要用来解决生产者消费者的问题，而且这个思想被很多应用借鉴，例如消息队列。
5.  生产者消费者虽然是一种面向多线程的模式，但是由于加锁操作限制了程序影响了整个程序的吞吐，所以我们可以将程序设计成无锁的模式，可以借鉴Disruptor

[img]: /images/20221021/5c73d381ef9643bd9e59f72f30594fc4.png
[image-20210502112802102]: /images/20221021/4bdaff45983541b7ac157336e7f00513.png
[image-20210502113258553]: /images/20221021/72f6660b2ddd484a9b9c940625d90403.png