手写生产者/消费者模式（三种实现方式）

Love The Way You Lie 2021-11-02 07:02 440阅读 0赞

生产者消费者模式需要满足：

1）生产者生产数据到缓冲区中，消费者从缓冲区中取数据；

2）缓冲区蛮，生产者线程阻塞；

3）缓冲区为空，消费者线程阻塞；

分析

1）定义缓存队列，选择一个集合当作缓存，给与缓存上线，缓存队列只有两种行为（生产数据和消费数据）。

2）定义生产者线程，调用缓存队列中的生产行为；

3）定义消费者线程，调用缓存队列中的消费者行为

第一种方式：

双向链表LinkedHashMap和Synchronized结合

定义缓存队列

> 1.  package pratice810;
> 2.  import java.util.Iterator;
> 3.  import java.util.LinkedHashMap;
> 4.  import java.util.Map;
> 5.  public class QueueSX<T> \{
> 6.  //公共缓存队列，1）生产，2）消费
> 7.      private int putIndex = 0;//数据插入的角标
> 8.      private int maxCount = 50;//缓存区最大长度
> 9.      private LinkedHashMap<Integer,T> lhm = new LinkedHashMap<Integer,T>();
> 10.     
> 11.     public synchronized void add(T msg)\{
> 12.         if(lhm.size() == maxCount)\{
> 13.         //如果缓存区达到最大数量，则阻塞生产者线程
> 14.             try \{
> 15.                 wait();
> 16.             \} catch (InterruptedException e) \{
> 17.                 // TODO Auto-generated catch block
> 18.                 e.printStackTrace();
> 19.             \}            
> 20.         \}else\{
> 21.             notifyAll();//唤醒所有线程
> 22.         \}
> 23.         
> 24.         lhm.put(putIndex, msg);
> 25.         putIndex = (putIndex+1 >= maxCount)?(putIndex+1) % maxCount:putIndex+1;        
> 26.     \}
> 27.     public synchronized T remove()\{
> 28.         if(lhm.size() == 0)\{
> 29.             try \{
> 30.                 //如果没有数据，阻塞消费者线程
> 31.                 wait();
> 32.             \} catch (InterruptedException e) \{
> 33.                 // TODO Auto-generated catch block
> 34.                 e.printStackTrace();
> 35.             \}
> 36.         \}else\{
> 37.             notifyAll();
> 38.         \}
> 39.         Iterator it = lhm.entrySet().iterator();
> 40.         T t = null;
> 41.         if(it.hasNext())\{
> 42.             Map.Entry<Integer, T> entry = (Map.Entry<Integer, T>)it.next();
> 43.             t = entry.getValue();
> 44.             int index = entry.getKey();
> 45.             lhm.remove(index);            
> 46.         \}
> 47.         return t;
> 48.     \}
> 49. \}

生产者

> 1.  package pratice810;
> 2.  //生产者线程
> 3.  public class productThread extends Thread \{
> 4.      private QueueSX queuesx;
> 5.      public productThread(QueueSX queuesx)\{
> 6.          this.queuesx = queuesx;
> 7.      \}
> 8.      @Override
> 9.      public void run()\{
> 10.         for(int i=0;i<60;i++)\{
> 11.             QueueSX.add(String.valueOf(i));
> 12.         \}
> 13.     \}
> 14. \}

消费者

> 1.  package pratice810;
> 2.  //消费者线程
> 3.  public class ConsumerThread extends Thread\{
> 4.      private QueueSX queuesx;
> 5.      public ConsumerThread(QueueSX queuesx)\{
> 6.          this.queuesx =queuesx;
> 7.      \}
> 8.      public void run()\{
> 9.          for(;;)
> 10.             queuesx.remove();
> 11.     \}
> 12. \}

启动：

> 1.  package pratice810;
> 2.  public class productConsumerTest \{
> 3.  //启动
> 4.      public static void main(String\[\] args) \{
> 5.          QueueSX queuesx = new QueueSX();
> 6.          productThread pt = new productThread(queuesx);
> 7.          ConsumerThread ct = new ConsumerThread(queuesx);
> 8.          pt.start();//启动线程
> 9.          ct.start();
> 10.     \}
> 11. \}

1）两个互斥线程，一个负责生产一个负责消费，线程不安全；

2）LinkedHashMap作为缓存队列，it.next()取出第一个数据，可以保证遍历的顺序；

3）为了保证互斥线程的安全性，需要做对应的处理，使用synchronized、wait()、notifyAll()来保证；

第二种方式

LinkedHashMap双向链表与lock结合。

private int putIndex = 0;//数据插入的角标  
    private int maxCount = 50;//缓存区最大长度  
    private Lock lock;  
    private Condition addCondition;  
    private Condition removeCondition;

public PublicQueue()\{  
        lock = new ReentrantLock();  
        addCondition = lock.newCondition();  
        removeCondition =lock.newCondition();  
    \}  
第三种

直接使用阻塞队列BlockingQueue

private BlockingDeque<T> blockingDeque = new LinkedBlockingDeque<>(50);//缓冲区

转载：[https://blog.csdn.net/Virgil\_K2017/article/details/89283946][https_blog.csdn.net_Virgil_K2017_article_details_89283946]

[https_blog.csdn.net_Virgil_K2017_article_details_89283946]: https://blog.csdn.net/Virgil_K2017/article/details/89283946