资讯专栏INFORMATION COLUMN

上云采购季！| 2核2G4M爆款云服务器低至59元/年，更有多台、长期优惠，快来选购！

立即前往
首页/文章专栏/实现生产者消费者模式的四种方式（Synchronized、Lock、Semaphore、Blocki

实现生产者消费者模式的四种方式（Synchronized、Lock、Semaphore、Blocki

godiscoder / 3279人阅读

摘要：所谓生产者消费者模式，即个线程进行生产，同时个线程进行消费，两种角色通过内存缓冲区进行通信图片来源下面我们通过四种方式，来实现生产者消费者模式。通过生产者调用，减少数目可以消费的数量。

所谓生产者消费者模式，即N个线程进行生产，同时N个线程进行消费，两种角色通过内存缓冲区进行通信

图片来源https://www.cnblogs.com/chent...

下面我们通过四种方式，来实现生产者消费者模式。

首先是最原始的synchronized方式

定义库存类（即图中缓存区）

</>复制代码 
  1. class Stock {
  2.     private String name;
  3.     // 标记库存是否有内容
  4.     private boolean hasComputer = false;
  5.     public synchronized void putOne(String name) {
  6.         // 若库存中已有内容，则生产线程阻塞等待
  7.         while (hasComputer) {
  8.             try {
  9.                 this.wait();
  10.             } catch (InterruptedException e) {
  11.                 e.printStackTrace();
  12.             }
  13.         }
  14.         this.name = name;
  15.         System.out.println("生产者...生产了 " + name);
  16.         // 更新标记
  17.         this.hasComputer = true;
  18.         // 这里用notify的话，假设p0执行完毕，此时c0,c1都在wait, 同时唤醒另一个provider:p1，
  19.         // p1判断标记后休眠，造成所有线程都wait的局面，即死锁；
  20.         // 因此使用notifyAll解决死锁问题
  21.         this.notifyAll();
  22.     }
  23.     public synchronized void takeOne() {
  24.         // 若库存中没有内容，则消费线程阻塞等待生产完毕后继续
  25.         while (!hasComputer) {
  26.             try {
  27.                 this.wait();
  28.             } catch (InterruptedException e) {
  29.                 e.printStackTrace();
  30.             }
  31.         }
  32.         System.out.println("消费者...消费了 " + name);
  33.         this.hasComputer = false;
  34.         this.notifyAll();
  35.     }
  36. }

定义生产者和消费者（为了节省空间和方便阅读，这里将生产者和消费者定义成了匿名内部类）

</>复制代码 
  1. public static void main(String[] args) {
  2.     // 用于通信的库存类
  3.     Stock computer = new Stock();
  4.     // 定义两个生产者和两个消费者
  5.     Thread p1 = new Thread(new Runnable() {
  6.         @Override
  7.         public void run() {
  8.             while (true) {
  9.                 computer.putOne("Dell");
  10.             }
  11.         }
  12.     });
  13.     Thread p2 = new Thread(new Runnable() {
  14.         @Override
  15.         public void run() {
  16.             while (true) {
  17.                 computer.putOne("Mac");
  18.             }
  19.         }
  20.     });
  21.     
  22.     Thread c1 = new Thread(new Runnable() {
  23.         @Override
  24.         public void run() {
  25.             while (true) {
  26.                 computer.takeOne();
  27.             }
  28.         }
  29.     });
  30.     Thread c2 = new Thread(new Runnable() {
  31.         @Override
  32.         public void run() {
  33.             while (true) {
  34.                 computer.takeOne();
  35.             }
  36.         }
  37.     });
  38.     p1.start();
  39.     p2.start();
  40.     c1.start();
  41.     c2.start();
  42. }

运行结果图

第二种方式：Lock

Jdk1.5之后加入了Lock接口，一个lock对象可以有多个Condition类,Condition类负责对lock对象进行wait,notify,notifyall操作

定义库存类

</>复制代码 
  1. class LockStock {
  2.     final Lock lock = new ReentrantLock();
  3.     final Condition notFull = lock.newCondition();
  4.     final Condition notEmpty = lock.newCondition();
  5.     // 加入库存概念，可批量生产和消费
  6.     // 定义最大库存为10
  7.     final String[] stock = new String[10];
  8.     // 写入标记、读取标记、已有商品数量
  9.     int putptr, takeptr, count;
  10.     public void put(String computer) {
  11.         // lock代替synchronized
  12.         lock.lock();
  13.         try {
  14.             // 若库存已满则生产者线程阻塞
  15.             while (count == stock.length)
  16.                 notFull.await();
  17.             // 库存中加入商品
  18.             stock[putptr] = computer;
  19.             // 库存已满，指针置零，方便下次重新写入
  20.             if (++putptr == stock.length) putptr = 0;
  21.             ++count;
  22.             System.out.println(computer + " 正在生产数据： -- 库存剩余：" + count);
  23.             notEmpty.signal();
  24.         } catch (InterruptedException e) {
  25.             e.printStackTrace();
  26.         } finally {
  27.             lock.unlock();
  28.         }
  29.         try {
  30.             Thread.sleep(1000);
  31.         } catch (InterruptedException e) {
  32.             e.printStackTrace();
  33.         }
  34.     }
  35.     public String take(String consumerName) {
  36.         lock.lock();
  37.         try {
  38.             while (count == 0)
  39.                 notEmpty.await();
  40.             // 从库存中获取商品
  41.             String computer = stock[takeptr];
  42.             if (++takeptr == stock.length) takeptr = 0;
  43.             --count;
  44.             System.out.println(consumerName + " 正在消费数据：" + computer + " -- 库存剩余：" + count);
  45.             notFull.signal();
  46.             return computer;
  47.         } catch (InterruptedException e) {
  48.             e.printStackTrace();
  49.         } finally {
  50.             lock.unlock();
  51.         }
  52.         // 无逻辑作用，放慢速度
  53.         try {
  54.             Thread.sleep(1000);
  55.         } catch (InterruptedException e) {
  56.             e.printStackTrace();
  57.         }
  58.         return "";
  59.     }
  60. }

以上部分代码摘自java7 API中Condition接口的官方示例

接着还是定义生产者和消费者

</>复制代码 
  1. public static void main(String[] args) {
  2.     LockStock computer = new LockStock();
  3.     Thread p1 = new Thread(new Runnable() {
  4.         @Override
  5.         public void run() {
  6.             while (true) {
  7.                 computer.put("Dell");
  8.             }
  9.         }
  10.     });
  11.     Thread p2 = new Thread(new Runnable() {
  12.         @Override
  13.         public void run() {
  14.             while (true) {
  15.                 computer.put("Mac");
  16.             }
  17.         }
  18.     });
  19.     Thread c1 = new Thread(new Runnable() {
  20.         @Override
  21.         public void run() {
  22.             while (true) {
  23.                 computer.take("zhangsan");
  24.             }
  25.         }
  26.     });
  27.     Thread c2 = new Thread(new Runnable() {
  28.         @Override
  29.         public void run() {
  30.             while (true) {
  31.                 computer.take("李四");
  32.             }
  33.         }
  34.     });
  35.     // 两个生产者两个消费者同时运行
  36.     p1.start();
  37.     p2.start();
  38.     c1.start();
  39.     c2.start();
  40. }

运行结果图：

第三种方式：Semaphore
首先依旧是库存类：

</>复制代码 
  1. class Stock {
  2.     List stock = new LinkedList();
  3.     // 互斥量，控制共享数据的互斥访问
  4.     private Semaphore mutex = new Semaphore(1);
  5.     // canProduceCount可以生产的总数量。 通过生产者调用acquire，减少permit数目
  6.     private Semaphore canProduceCount = new Semaphore(10);
  7.     // canConsumerCount可以消费的数量。通过生产者调用release，增加permit数目
  8.     private Semaphore canConsumerCount = new Semaphore(0);
  9.     public void put(String computer) {
  10.         try {
  11.             // 可生产数量 -1
  12.             canProduceCount.acquire();
  13.             mutex.acquire();
  14.             // 生产一台电脑
  15.             stock.add(computer);
  16.             System.out.println(computer + " 正在生产数据" + " -- 库存剩余：" + stock.size());
  17.         } catch (InterruptedException e) {
  18.             e.printStackTrace();
  19.         } finally {
  20.             // 释放互斥锁
  21.             mutex.release();
  22.             // 释放canConsumerCount，增加可以消费的数量
  23.             canConsumerCount.release();
  24.         }
  25.         // 无逻辑作用，放慢速度
  26.         try {
  27.             Thread.sleep(500);
  28.         } catch (InterruptedException e) {
  29.             e.printStackTrace();
  30.         }
  31.     }
  32.     public void get(String consumerName) {
  33.         try {
  34.             // 可消费数量 -1
  35.             canConsumerCount.acquire();
  36.             mutex.acquire();
  37.             // 从库存消费一台电脑
  38.             String removedVal = stock.remove(0);
  39.             System.out.println(consumerName + " 正在消费数据：" + removedVal + " -- 库存剩余：" + stock.size());
  40.         } catch (InterruptedException e) {
  41.             e.printStackTrace();
  42.         } finally {
  43.             mutex.release();
  44.             // 消费后释放canProduceCount，增加可以生产的数量
  45.             canProduceCount.release();
  46.         }
  47.     }
  48. }

还是生产消费者：

</>复制代码 
  1. public class SemaphoreTest {
  2.     public static void main(String[] args) {
  3.         // 用于多线程操作的库存变量
  4.         final Stock stock = new Stock();
  5.         // 定义两个生产者和两个消费者
  6.         Thread dellProducer = new Thread(new Runnable() {
  7.             @Override
  8.             public void run() {
  9.                 while (true) {
  10.                     stock.put("Del");
  11.                 }
  12.             }
  13.         });
  14.         Thread macProducer = new Thread(new Runnable() {
  15.             @Override
  16.             public void run() {
  17.                 while (true) {
  18.                     stock.put("Mac");
  19.                 }
  20.             }
  21.         });
  22.         Thread consumer1 = new Thread(new Runnable() {
  23.             @Override
  24.             public void run() {
  25.                 while (true) {
  26.                     stock.get("zhangsan");
  27.                 }
  28.             }
  29.         });
  30.         Thread consumer2 = new Thread(new Runnable() {
  31.             @Override
  32.             public void run() {
  33.                 while (true) {
  34.                     stock.get("李四");
  35.                 }
  36.             }
  37.         });
  38.         dellProducer.start();
  39.         macProducer.start();
  40.         consumer1.start();
  41.         consumer2.start();
  42.     }
  43. }

运行结果图：

第四种方式：BlockingQueue
BlockingQueue的put和take底层实现其实也是使用了第二种方式中的ReentrantLock+Condition，并且帮我们实现了库存队列，方便简洁
1、定义生产者

</>复制代码 
  1. class Producer implements Runnable {
  2.     // 库存队列
  3.     private BlockingQueue stock;
  4.     // 生产/消费延迟
  5.     private int timeOut;
  6.     private String name;
  7.     public Producer(BlockingQueue stock, int timeout, String name) {
  8.         this.stock = stock;
  9.         this.timeOut = timeout;
  10.         this.name = name;
  11.     }
  12.     @Override
  13.     public void run() {
  14.         while (true) {
  15.             try {
  16.                 stock.put(name);
  17.                 System.out.println(name + " 正在生产数据" + " -- 库存剩余：" + stock.size());
  18.                 TimeUnit.MILLISECONDS.sleep(timeOut);
  19.             } catch (InterruptedException e) {
  20.                 e.printStackTrace();
  21.             }
  22.         }
  23.     }
  24. }

2、定义消费者

</>复制代码 
  1. class Consumer implements Runnable {
  2.     // 库存队列
  3.     private BlockingQueue stock;
  4.     private String consumerName;
  5.     public Consumer(BlockingQueue stock, String name) {
  6.         this.stock = stock;
  7.         this.consumerName = name;
  8.     }
  9.     @Override
  10.     public void run() {
  11.         while (true) {
  12.             try {
  13.                 // 从库存消费一台电脑
  14.                 String takeName = stock.take();
  15.                 System.out.println(consumerName + " 正在消费数据：" + takeName + " -- 库存剩余：" + stock.size());
  16.             } catch (InterruptedException e) {
  17.                 e.printStackTrace();
  18.             }
  19.         }
  20.     }
  21. }

3、定义库存并运行

</>复制代码 
  1. public static void main(String[] args) {
  2.         // 定义最大库存为10
  3.         BlockingQueue stock = new ArrayBlockingQueue<>(10);
  4.         Thread p1 = new Thread(new Producer(stock, 500, "Mac"));
  5.         Thread p2 = new Thread(new Producer(stock, 500, "Dell"));
  6.         Thread c1 = new Thread(new Consumer(stock,"zhangsan"));
  7.         Thread c2 = new Thread(new Consumer(stock, "李四"));
  8.         p1.start();
  9.         p2.start();
  10.         c1.start();
  11.         c2.start();
  12.     }

运行结果图：

感谢阅读~欢迎指正和补充~~~

云服务器 GPU云服务器 对数据仓库分析的四种方式 对集群生产消费实现数据迁移 activity的四种状态 数据迁移的四种方法

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/76979.html

相关文章

  • Java 中的 Monitor 机制

    摘要：基本元素机制需要几个元素来配合，分别是临界区对象及锁条件变量以及定义在对象上的，操作。这个外部条件在机制中称为条件变量。提供的机制，其实是，等元素合作形成的，甚至说外部的条件变量也是个组成部分。 monitor的概念 管程，英文是 Monitor，也常被翻译为监视器，monitor 不管是翻译为管程还是监视器，都是比较晦涩的，通过翻译后的中文，并无法对 monitor 达到一个直观的描...

    Jacendfeng 评论0 收藏0

  • 想进大厂？50个多线程面试题，你会多少？（一）

    摘要：下面是线程相关的热门面试题，你可以用它来好好准备面试。线程安全问题都是由全局变量及静态变量引起的。持有自旋锁的线程在之前应该释放自旋锁以便其它线程可以获得自旋锁。 最近看到网上流传着，各种面试经验及面试题，往往都是一大堆技术题目贴上去，而没有答案。 不管你是新程序员还是老手，你一定在面试中遇到过有关线程的问题。Java语言一个重要的特点就是内置了对并发的支持，让Java大受企业和程序员...

    wow_worktile 评论0 收藏0

  • 40道阿里巴巴JAVA研发岗多线程面试题详解，你能答出多少

    摘要：但是单核我们还是要应用多线程，就是为了防止阻塞。多线程可以防止这个问题，多条线程同时运行，哪怕一条线程的代码执行读取数据阻塞，也不会影响其它任务的执行。 1、多线程有什么用？一个可能在很多人看来很扯淡的一个问题：我会用多线程就好了，还管它有什么用？在我看来，这个回答更扯淡。所谓知其然知其所以然，会用只是知其然，为什么用才是知其所以然，只有达到知其然知其所以然的程度才可以说是把一个知识点...

    lpjustdoit 评论0 收藏0

  • Java 并发学习笔记

    摘要：方法可以将当前线程放入等待集合中，并释放当前线程持有的锁。此后，该线程不会接收到的调度，并进入休眠状态。该线程会唤醒，并尝试恢复之前的状态。 并发 最近重新复习了一边并发的知识，发现自己之前对于并发的了解只是皮毛。这里总结以下Java并发需要掌握的点。 使用并发的一个重要原因是提高执行效率。由于I/O等情况阻塞，单个任务并不能充分利用CPU时间。所以在单处理器的机器上也应该使用并发。为...

    DrizzleX 评论0 收藏0

发表评论

登陆后可评论

0条评论

godiscoder

男|高级讲师
我要私信

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<