常见场景:
某个模块负责产生数据,这些数据由另一个模块来负责处理。产生数据的模块,就形象地称为生产者;而处理数据的模块,就称为消费者。
该模式还需要有一个缓冲区处于生产者和消费者之间,作为一个中介。生产者把数据放入缓冲区,而消费者从缓冲区取出数据
缓冲区作用
1. 解耦,生产者和消费者只依赖缓冲区,而不互相依赖
2. 支持并发和异步
方式一,同步队列
/** * 生产者、消费者缓冲区 */ public class Storage implements IStorage { private final int maxSize = 10; private Queue<Object> queue = new LinkedList<Object>(); @Override public void put(Object obj) { synchronized (queue) { while (queue.size() > maxSize) { System.out.println("缓冲区已满,不能进入"); try { queue.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } queue.add(obj); System.out.println("进入缓冲区"); queue.notifyAll(); } } @Override public Object get() { Object obj = null; synchronized (queue) { while (queue.size() <= 0) { System.out.println("缓冲区为空, 进入等待"); try { queue.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } obj = queue.poll(); System.out.println("离开缓冲区"); queue.notifyAll(); } return obj; } }
方式二,可重入锁
public class Storage implements IStorage { private final int maxSize = 20; private LinkedList<Object> list = new LinkedList<Object>(); private final Lock lock = new ReentrantLock(); // 仓库满的条件变量 private final Condition full = lock.newCondition(); // 仓库空的条件变量 private final Condition empty = lock.newCondition(); @Override public void put(Object obj) { lock.lock(); while (list.size() >= maxSize) { try { System.out.println("缓冲区已满,不能进入"); // 生产阻塞 full.await(); } catch (InterruptedException e) { e.printStackTrace(); } } list.add(obj); System.out.println("进入缓冲区"); empty.signalAll(); lock.unlock(); } @Override public Object get() { lock.lock(); while (list.size() <= 0) { try { System.out.println("缓冲区为空, 进入等待"); // 消费阻塞 empty.await(); } catch (InterruptedException e) { e.printStackTrace(); } } Object obj = list.remove(); System.out.println("离开缓冲区"); full.signalAll(); lock.unlock(); return obj; } }
方式三,阻塞队列
public class Storage implements IStorage { private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<Object>(10); @Override public void put(Object obj) { try { list.put(obj); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("进入缓冲区"); } @Override public Object get() { Object obj = null; try { obj = list.take(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("离开缓冲区"); return obj; } }