概述

生产者消费者模式是一个多线程协作的模式，在这个模式中，一部分线程负责生产数据、一部分线程负责消费数据，通过缓冲区完成生产者与消费者的解耦，缓冲区相当于数据仓库。
image.png

实现方式

wait/notify

  1. //数据仓库，缓冲区
  2. public class DataBuffer {
  4.     List<String> dataBuffer = new LinkedList<>();
  6.     public void add(String data){
  7.         dataBuffer.add(data);
  8.     }
  10.     public String get(){
  11.         String data = null;
  12.         if(dataBuffer.size() > 0){
  13.             data = dataBuffer.get(0);
  14.             if(data != null){
  15.                 dataBuffer.remove(data);
  16.             }
  17.         }
  18.         return data;
  19.     }
  20. }
  22. //生产者
  23. public class Producter implements Runnable {
  25.     private DataBuffer dataBuffer;
  26.     AtomicInteger atomicInteger = new AtomicInteger(0);
  28.     public Producter(DataBuffer dataBuffer){
  29.         this.dataBuffer = dataBuffer;
  30.     }
  32.     @Override
  33.     public void run() {
  34.         while (true){
  35.             synchronized (dataBuffer){
  36.                 atomicInteger.incrementAndGet();
  37.                 dataBuffer.add("a" + atomicInteger.intValue());
  38.                 System.out.println("生产数据，并唤醒等待线程");
  39.                 dataBuffer.notifyAll();
  40.                 try {
  41.                     //Thread.sleep(100L);
  42.                 }catch (Exception e) {
  44.                 }
  45.             }
  46.         }
  47.     }
  48. }
  51. //消费者
  52. public class Consumer implements Runnable {
  53.     private DataBuffer dataBuffer;
  55.     public Consumer(DataBuffer dataBuffer){
  56.         this.dataBuffer = dataBuffer;
  57.     }
  59.     @Override
  60.     public void run() {
  61.         while (true){
  62.             synchronized (dataBuffer){
  63.                 String a = dataBuffer.get();
  64.                 if(a == null){
  65.                     try {
  66.                         dataBuffer.wait();
  67.                     }catch (Exception e){
  68.                         e.printStackTrace();
  69.                     }
  70.                 }else {
  71.                     System.out.println("消费的数据为：" + a);
  72.                 }
  73.             }
  74.         }
  75.     }
  76. }
  78. //测试类
  79. public class TestOne {
  80.     public static void main(String[] args){
  81.         DataBuffer dataBuffer = new DataBuffer();
  82.         new Thread(new Producter(dataBuffer)).start();
  83.         new Thread(new Consumer(dataBuffer)).start();
  84.         new Thread(new Producter(dataBuffer)).start();
  85.         new Thread(new Consumer(dataBuffer)).start();
  86.         new Thread(new Producter(dataBuffer)).start();
  87.         new Thread(new Consumer(dataBuffer)).start();
  88.     }
  89. }

BlockingQueue

  1. //数据缓冲区
  2. public class DataBuffer {
  4.     BlockingQueue<String> dataBuffer = new LinkedBlockingQueue<>(20);
  5.     public void add(String data){
  6.         try {
  7.             dataBuffer.put(data);
  8.         }catch (Exception e){
  10.         }
  11.     }
  13.     public String get(){
  14.         String data = null;
  15.         try {
  16.             data = dataBuffer.take();
  17.         }catch (Exception e){
  19.         }
  20.         return data;
  21.     }
  22. }
  23. //生产者
  24. public class Producter implements Runnable {
  26.     private DataBuffer dataBuffer;
  27.     AtomicInteger atomicInteger = new AtomicInteger(0);
  29.     public Producter(DataBuffer dataBuffer){
  30.         this.dataBuffer = dataBuffer;
  31.     }
  33.     @Override
  34.     public void run() {
  35.         int b = atomicInteger.incrementAndGet();
  36.         dataBuffer.add("a" + b);
  37.         System.out.println("生产数据 - a" + b);
  38.     }
  39. }
  41. //消费者
  42. public class Consumer implements Runnable {
  43.     private DataBuffer dataBuffer;
  45.     public Consumer(DataBuffer dataBuffer){
  46.         this.dataBuffer = dataBuffer;
  47.     }
  49.     @Override
  50.     public void run() {
  51.         while (true){
  52.             String a = dataBuffer.get();
  53.             System.out.println("消费的数据为：" + a);
  54.         }
  55.     }
  56. }
  58. //测试类
  59. public class TestTwo {
  61.     public static void main(String[] args){
  62.         DataBuffer dataBuffer = new DataBuffer();
  63.         new Thread(new Producter(dataBuffer)).start();
  64.         new Thread(new Consumer(dataBuffer)).start();
  65.         new Thread(new Producter(dataBuffer)).start();
  66.         new Thread(new Consumer(dataBuffer)).start();
  67.         new Thread(new Producter(dataBuffer)).start();
  68.         new Thread(new Consumer(dataBuffer)).start();
  69.     }
  70. }

ReentrantLock&Condition

  1. public class DataBuffer {
  3.     List<String> dataBuffer = new LinkedList<>();
  4.     Lock lock = new ReentrantLock();
  5.     Condition condition = lock.newCondition();
  7.     public Lock getLock(){
  8.         return lock;
  9.     }
  11.     public void add(String data){
  12.         dataBuffer.add(data);
  13.     }
  15.     public String get(){
  16.         String data = null;
  17.         if(dataBuffer.size() > 0){
  18.             data = dataBuffer.get(0);
  19.             if(data != null){
  20.                 dataBuffer.remove(data);
  21.             }
  23.         }
  25.         return data;
  26.     }
  27. }
  30. public class Producter implements Runnable {
  32.     private DataBuffer dataBuffer;
  33.     AtomicInteger atomicInteger = new AtomicInteger(0);
  35.     public Producter(DataBuffer dataBuffer){
  36.         this.dataBuffer = dataBuffer;
  37.     }
  39.     @Override
  40.     public void run() {
  41.         while (true){
  42.             Lock lock = dataBuffer.getLock();
  43.             try {
  44.                 lock.lock();
  45.                 atomicInteger.incrementAndGet();
  46.                 dataBuffer.add("a" + atomicInteger.intValue());
  47.                 System.out.println("生产数据，并唤醒等待线程");
  48.                 dataBuffer.condition.signalAll();
  49.                 //Thread.sleep(100L);
  50.             }catch (Exception e) {
  52.             }finally {
  53.                 lock.unlock();
  54.             }
  55.         }
  56.     }
  57. }
  60. public class Consumer implements Runnable {
  61.     private DataBuffer dataBuffer;
  63.     public Consumer(DataBuffer dataBuffer){
  64.         this.dataBuffer = dataBuffer;
  65.     }
  67.     @Override
  68.     public void run() {
  69.         while (true){
  70.             Lock lock = dataBuffer.getLock();
  71.             try {
  72.                 lock.lock();
  73.                 String a = dataBuffer.get();
  74.                 if(a == null){
  75.                     try {
  76.                         dataBuffer.condition.await();
  77.                     }catch (Exception e){
  78.                         e.printStackTrace();
  79.                     }
  80.                 }else {
  81.                     System.out.println("消费的数据为：" + a);
  82.                 }
  83.             }catch (Exception e){
  85.             }finally {
  86.                 lock.unlock();
  87.             }
  88.         }
  89.     }
  90. }
  93. public class TestOne {
  95.     public static void main(String[] args){
  96.         DataBuffer dataBuffer = new DataBuffer();
  97.         new Thread(new Producter(dataBuffer)).start();
  98.         new Thread(new Consumer(dataBuffer)).start();
  99.         new Thread(new Producter(dataBuffer)).start();
  100.         new Thread(new Consumer(dataBuffer)).start();
  101.         new Thread(new Producter(dataBuffer)).start();
  102.         new Thread(new Consumer(dataBuffer)).start();
  103.     }
  104. }