生产者和消费者三种实现:wait/notify、lock、disruptor

淡淡的烟草味﹌ 2022-02-23 05:58 211阅读 0赞

# 第一种 wait/notify+链表实现 #

生产者

/** * @author sunchangtan * @date 2019/2/25 12:59 */
    public class Producer implements Runnable { 
        private final List<Goods> goods;
        private int maxCapiblity;
    
        public Producer(List<Goods> goods, int maxCapiblity) { 
            this.goods = goods;
            this.maxCapiblity = maxCapiblity;
        }
    
        @Override
        public void run() { 
            while (true) { 
                synchronized (goods) {  //要使用goods或者全局锁对象，不能用this，因为对象头中的monitor必须要相同一个
                    while (goods.size() >= maxCapiblity) {  // 必须使用while，避免线程假唤醒，因为wait操作有三个步骤，而且wait会释放锁，需要理解假唤醒的原因
                        try { 
                            System.out.println("仓库已满，等待消费");
                            goods.wait();  //需要同一个monitor，否则将会线程一直waiting
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
    
                    Goods newGoods = this.produceNewGoods();
                    this.goods.add(newGoods);
                    System.out.println("生产商品" + newGoods.getGoodsId());
                    goods.notifyAll();  //需要同一个monitor，否则将会该线程不可能会唤醒其他Consumer线程。需要理解monitor作用
                }
            }
        }
    
        private Goods produceNewGoods() { 
            Goods goods = new Goods();
            goods.setGoodsId(UUID.randomUUID().toString());
            goods.setGoodsName(goods.getGoodsId());
            goods.setPrice(new BigDecimal(10));
            goods.setQuantity(1);
            return goods;
        }
    }

消费者

/** * wait/notfiy实现生产者和消费者 * * @author sunchangtan * @date 2019/2/25 13:23 */
    public class Consumer implements Runnable { 
        private final List<Goods> goods;
    
        public Consumer(List<Goods> goods) { 
            this.goods = goods;
        }
    
        @Override
        public void run() { 
            while (true) { 
                synchronized (goods) {  //要使用goods或者全局锁对象，不能用this，因为对象头中的monitor必须要相同一个。 Consumer和Producer操作goods不在一起
                    while (goods.size() == 0) {  // 必须使用while，避免线程假唤醒，因为wait操作有三个步骤，而且wait会释放锁，需要理解假唤醒的原因
                        try { 
                            System.out.println("没有商品，等待生产");
                            goods.wait();  //需要同一个monitor
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
    
                    Goods goodsToConsume = this.goods.remove(0);
                    System.out.println("消费商品" + goodsToConsume.getGoodsId());
                    goods.notifyAll(); //需要同一个monitor，否则将会该线程不可能会唤醒其他Producer线程。需要理解monitor作用
                }
            }
    
        }
    }

# 第二种wait/notify + 环形数组容器对象 #

定义环形数组容器

/** * @author sunchangtan * @date 2019/2/26 14:03 */
    public class BasicGoodsWarehouse { 
        //环形数组
        private int maxCapacity;
        private Goods[] goodsArr;
        private int head;
        private int tail;
    
        public BasicGoodsWarehouse(int maxCapacity) { 
            this.maxCapacity = maxCapacity;
            goodsArr = new Goods[this.maxCapacity];
            head = 0;
            tail = 0;
        }
    
        /** * 向GoodsWarehouse生产商品 * @param newGoods */
        public void offer(Goods newGoods) { 
            synchronized (this) {  //要可以用this，因为GoodsWarehouse对象头中的monitor可以生产和消费共享
                while ((tail > head) ? (tail - head + 1 >= maxCapacity) : (head - tail - 1 >= 0)) {  //使用循环数组，减少内存分配，而ArrayList是需要拷贝的
                    try { 
                        this.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
    
    
                this.goodsArr[tail++] = newGoods;
                if(tail == maxCapacity - 1) { 
                    tail = 0;
                }
                this.notifyAll(); //要可以用this，因为GoodsWarehouse对象头中的monitor可以生产和消费共享的
            }
        }
    
        /** * 从GoodsWarehouse中消费商品 * @return */
        public Goods take() { 
            synchronized (this) { 
                while (tail == head) { 
                    try { 
                        this.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
    
                Goods goods = this.goodsArr[head++];
                if(head == maxCapacity - 1) { 
                    head = 0;
                }
                this.notifyAll();
                return goods;
            }
        }
    }

生产者

/** * @author sunchangtan * @date 2019/2/26 15:52 */
    public class Producer implements Runnable { 
        private BasicGoodsWarehouse goodsWarehouse;
    
        public Producer(BasicGoodsWarehouse goodsWarehouse) { 
            this.goodsWarehouse = goodsWarehouse;
        }
    
        @Override
        public void run() { 
            while (true) { 
                Goods newGoods = new Goods();
                newGoods.setGoodsId(UUID.randomUUID().toString());
    
                goodsWarehouse.offer(newGoods);
    
                System.out.println(Thread.currentThread().getName() + "生产商品: " + newGoods.getGoodsId());
    
    
                try { 
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

消费者

/** * @author sunchangtan * @date 2019/2/26 15:55 */
    public class Consumer implements Runnable { 
        private BasicGoodsWarehouse goodsWarehouse;
    
        public Consumer(BasicGoodsWarehouse goodsWarehouse) { 
            this.goodsWarehouse = goodsWarehouse;
        }
    
    
        @Override
        public void run() { 
            while (true) { 
                Goods goods = this.goodsWarehouse.take();
    
                System.out.println(Thread.currentThread().getName() + "消费商品: " + goods.getGoodsId());
    
                try { 
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

# 第三种 ReentrantLock+环形数组 #

环形数组容器

**
     * @author sunchangtan
     * @date 2019/2/26 14:03
     */
    public class LockGoodsWarehouse { 
        //环形数组
        private int maxCapacity;
        private Goods[] goodsArr;
        private int head;
        private int tail;
    
        private Lock lock = new ReentrantLock();
        private Condition notfullCondition = lock.newCondition();
        private Condition notEmptyCondition = lock.newCondition();
    
        public LockGoodsWarehouse(int maxCapacity) { 
            this.maxCapacity = maxCapacity;
            goodsArr = new Goods[this.maxCapacity];
            head = 0;
            tail = 0;
        }
    
        /** * 向GoodsWarehouse生产商品 * * @param newGoods */
        public void offer(Goods newGoods) { 
            lock.lock();
            try { 
                while ((tail > head) ? (tail - head + 1 >= maxCapacity) : (head - tail - 1 >= 0)) { 
                    try { 
                        notfullCondition.await();
                    } catch (InterruptedException e) { 
                        Thread.currentThread().interrupt();
                    }
                }
    
                this.goodsArr[tail++] = newGoods;
                if (tail == maxCapacity - 1) { 
                    tail = 0;
                }
                notEmptyCondition.signalAll();
            } finally { 
                lock.unlock();
            }
    
        }
    
        /** * 从GoodsWarehouse中消费商品 * * @return */
        public Goods take() { 
            lock.lock();
            try { 
                while (tail == head) { 
                    try { 
                        notEmptyCondition.await();
                    } catch (InterruptedException e) { 
                        Thread.currentThread().interrupt();
                    }
                }
    
                Goods goods = this.goodsArr[head++];
                if (head == maxCapacity - 1) { 
                    head = 0;
                }
                notfullCondition.signalAll();
                return goods;
            } finally { 
                lock.unlock();
            }
    
        }
    }

生产者

/** * @author sunchangtan * @date 2019/2/26 15:52 */
    public class Producer implements Runnable { 
        private LockGoodsWarehouse goodsWarehouse;
    
        public Producer(LockGoodsWarehouse goodsWarehouse) { 
            this.goodsWarehouse = goodsWarehouse;
        }
    
        @Override
        public void run() { 
            while (true) { 
                Goods newGoods = new Goods();
                newGoods.setGoodsId(UUID.randomUUID().toString());
    
                goodsWarehouse.offer(newGoods);
    
                System.out.println(Thread.currentThread().getName() + "生产商品: " + newGoods.getGoodsId());
    
    
                try { 
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

消费者

/** * @author sunchangtan * @date 2019/2/26 15:55 */
    public class Consumer implements Runnable { 
        private LockGoodsWarehouse goodsWarehouse;
    
        public Consumer(LockGoodsWarehouse goodsWarehouse) { 
            this.goodsWarehouse = goodsWarehouse;
        }
    
    
        @Override
        public void run() { 
            while (true) { 
                Goods goods = this.goodsWarehouse.take();
    
                System.out.println(Thread.currentThread().getName() + "消费商品: " + goods.getGoodsId());
    
                try { 
                    Thread.sleep((long) (Math.random() * 1000));
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

# 第四种 Disruptor实现 #

定义容器

/** * 定义元素的事件 */
    @Data
    public class GoodsEvent { 
        private Goods goods;
    
        private long sequence;
    }
    
    /** * 定义事件生成者 */
    public class GoodsEventFactory implements EventFactory<GoodsEvent> { 
        @Override
        public GoodsEvent newInstance() { 
            return new GoodsEvent();
        }
    }
    
    /** * 具体消费对象转换为消费事件Event对象 */
    public class GoodsTranslator implements EventTranslatorOneArg<GoodsEvent, Goods> { 
        @Override
        public void translateTo(GoodsEvent event, long sequence, Goods data) { 
            event.setGoods(data);
            event.setSequence(sequence);
        }
    }
    
    /** * Disruptor实现生产者和消费者 * @author sunchangtan */
    public class DisruptorGoodsWarehouse { 
        private Disruptor<GoodsEvent> disruptor;
    
        private int eventHandlerCount = 0;
        private EventHandlerGroup<GoodsEvent> eventHandlerGroup;
    
        public DisruptorGoodsWarehouse(int bufferSize) { 
            ThreadFactory threadFactory = new ThreadFactory() { 
                private final AtomicInteger index = new AtomicInteger(1);
    
                @Override
                public Thread newThread(Runnable r) { 
                    return new Thread(null, r, "disruptor-thread-" + index.getAndIncrement());
                }
            };
            
            //定义Disruptor 和 RingBuffer
            disruptor = new Disruptor<>(new GoodsEventFactory(), bufferSize, threadFactory, ProducerType.MULTI, new BlockingWaitStrategy());
        }
    
        /** * 增加消费者EventHandler，有消费依赖 * @param goodsEventHandler */
        public void addEventHandler(EventHandler<GoodsEvent>... goodsEventHandler) { 
            if (eventHandlerCount == 0) { 
                eventHandlerGroup = disruptor.handleEventsWith(goodsEventHandler);
            } else { 
                eventHandlerGroup.then(goodsEventHandler);
            }
            eventHandlerCount++;
        }
    
        public void start() { 
            if(disruptor == null) { 
                return;
            }
            disruptor.start();
        }
    
        public void shutdown() { 
            if(disruptor == null) { 
                return;
            }
            disruptor.shutdown();
        }
    
        public RingBuffer<GoodsEvent> getRingBuffer() { 
            return disruptor.getRingBuffer();
        }
    
    }

生产者

public class Producer implements Runnable { 
        private RingBuffer<GoodsEvent> ringBuffer;
    
        private static final GoodsTranslator TRANSLATOR = new GoodsTranslator();
    
        public Producer(RingBuffer<GoodsEvent> ringBuffer) { 
            this.ringBuffer = ringBuffer;
        }
    
        @Override
        public void run() { 
            while (true) { 
                Goods newGoods = new Goods();
                newGoods.setGoodsId(UUID.randomUUID().toString());
                ringBuffer.publishEvent(TRANSLATOR, newGoods);
    
                try { 
                   Thread.sleep(10000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
    
        }
    }

消费者

public class Consumer implements EventHandler<GoodsEvent> { 
        @Override
        public void onEvent(GoodsEvent event, long sequence, boolean endOfBatch) throws Exception { 
            System.out.println(Thread.currentThread().getName() + "==>" + event);
        }
    }