生产者消费者模式

2018-03-12  本文已影响0人  面包树_A

WHY

解决并发问题。

WHAT

通过一个容器解决生产者和消费者强耦合问题，否则消费者只能排队等生产者串行解决。
生产者和消费者之间不直接通信，而是通过一个阻塞队列来通信。阻塞队列就是一个缓冲区。

HOW

一、阻塞队列实现

阻塞队列是解决生产者消费者的关键。
-- 两种自定义阻塞队列方式
-- jdk1.5版本以后java.util.concurrent提供了阻塞队列类

阻塞队列接口：

package com.bytebeats.concurrent.queue;

/* 阻塞队列接口*/
public interface IBlockingQueue{
  public Object take();
  public void put(Object data);
}
方式1.推荐使用jdk提供的阻塞队列类

jdk1.5版本后新增的java.util.concurrent包新增了java.util.concurrent.BlockingQueue接口，并且提供了以下几种实现：

方式2.用Object.wai()/notifyAll() 实现自定义阻塞队列
package blocking.queue;

import java.util.LinkedList;

public class TraditionalBlockingQueue implements IBlockingQueue{
    
    /*这个表示该阻塞队列长度，最多能接受的数据数量*/
    private int queueSize;
    private final LinkedList<Object> queueList = new LinkedList<Object>();
    private final Object lock = new Object();

    public TraditionalBlockingQueue(int queueSize) {
        if (queueSize < 1) {
            throw new IllegalArgumentException("queueSize must be positive number");
        }
        this.queueSize = queueSize;
    }

    @Override
    public void put(Object data) throws InterruptedException {
        synchronized (lock) {
            /*这里必须使用while，而不使用if，这是一种自旋锁*/
            while (queueList.size() >= queueSize) {
                System.out.println("data is full");
                lock.wait();
            }
            queueList.add(data);
            lock.notifyAll();
        }
    }

    @Override
    public Object take() throws InterruptedException {
        synchronized (lock) {
            while(queueList.size() < 1) {
                System.out.println("data is empty");
                lock.wait();
            }
            Object data = queueList.removeFirst();
            /*此处必须使用notifyAll，如果使用notify在某些情况下会导致生产者-消费者同时处于阻塞情况*/
            lock.notifyAll();
            return data;
        }
    }
}
方式3.用Lock和Condition实现自定义阻塞队列

暂不考虑

二、生产者-消费者实现

生产者

package blocking.queue;

public class Producer implements Runnable {
    private IBlockingQueue queue;
    private int consumerNum;
    private int dataSize;

    public Producer(IBlockingQueue queue, int consumerNum, int dataSize) {
        this.queue = queue;
        this.dataSize = dataSize;
        this.consumerNum = consumerNum;
    }

    @Override
    public void run() {
        /*我这个Producer想往BlockingQueue里面放多少数据*/
        for (int i = 0; i < dataSize; i++) {
            try {
                queue.put("data_" + i);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        /*为每个Consumer创建一个结束标志位，是为了保证最后每个Consumer线程都能退出*/
        for (int i = 0; i < consumerNum; i++) {
            try {
                queue.put("#");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }   

        System.out.println("Producer over");
    }
}

消费者

package blocking.queue;

import java.util.concurrent.TimeUnit;

public class Consumer implements Runnable {

    private IBlockingQueue queue;

    public Consumer(IBlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            String data = null;
            try {
                data = (String) queue.take();
                System.out.println("Consumer " + Thread.currentThread().getName() + " consume:" + data);
                /*consumer领到#就退出，而#和consumer个数是一致的，所以每个consumer都能领到，然后安全退出。*/
                if ("#".equals(data)) {
                    break;
                }
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        System.out.println("Consumer over");
    }
}

测试代码

package blocking.queue;

public class QueueTest {

    public static void main(String[] args) {
        
        /*2表示该queue最多能放2个数据*/
        new QueueTest().testRun(new TraditionalBlockingQueue(2));

    }

    private void testRun(TraditionalBlockingQueue queue) {
        /*9表示有9个Consumer，是为了告诉Producer要设置多少个结束标志位（这是个比较土的通知Consumer线程结束的方式）
          20表示Producer想创建多少个数据（和Queue的区别是这20个是等Queue Not Full才分别放进去）*/
        Thread producer = new Thread(new Producer(queue, 9, 20));
        producer.start();

        Thread consumer1 = new Thread(new Consumer(queue));
        consumer1.start();
        Thread consumer2 = new Thread(new Consumer(queue));
        consumer2.start();
        Thread consumer3 = new Thread(new Consumer(queue));
        consumer3.start();
        Thread consumer4 = new Thread(new Consumer(queue));
        consumer4.start();
        Thread consumer5 = new Thread(new Consumer(queue));
        consumer5.start();
        Thread consumer6 = new Thread(new Consumer(queue));
        consumer6.start();
        Thread consumer7 = new Thread(new Consumer(queue));
        consumer7.start();
        Thread consumer8 = new Thread(new Consumer(queue));
        consumer8.start();
        Thread consumer9 = new Thread(new Consumer(queue));
        consumer9.start();
    }
}

测试结果

data is full
data is full
Consumer Thread-1 consume:data_0
Consumer Thread-2 consume:data_1
data is full
Consumer Thread-3 consume:data_2
data is full
Consumer Thread-4 consume:data_3
data is full
Consumer Thread-5 consume:data_4
data is full
Consumer Thread-6 consume:data_5
data is full
Consumer Thread-7 consume:data_6
data is full
Consumer Thread-8 consume:data_7
data is full
Consumer Thread-9 consume:data_8
data is full
data is full
Consumer Thread-4 consume:data_9
Consumer Thread-5 consume:#
Consumer over
data is full
Consumer Thread-6 consume:#
Consumer over
Consumer Thread-1 consume:#
Consumer over
Consumer Thread-2 consume:#
Consumer over
Consumer Thread-7 consume:#
Consumer over
data is empty
data is empty
data is full
Consumer Thread-3 consume:#
Consumer over
Consumer Thread-9 consume:#
Consumer over
Producer over
Consumer Thread-8 consume:#
Consumer over
Consumer Thread-4 consume:#
Consumer over

测试结果：Producer执行put的时候会校验queue是不是满了，满了要等不满再put，Consumer执行take的时候会校验queue是不是空的，空了要等不空再take。
每个consumer都会在遇到#后结束，并且consumer去抢夺producer产生的data。

上一篇下一篇

猜你喜欢

热点阅读