java消费者生产者模式及JDK之阻塞队列LinkedBlockingQueue实现
时间:2022-03-14 04:19
生产者消费者问题
(英语:Producer-consumer problem),也称有限缓冲问题(英语:Bounded-buffer problem),是一个问题的经典案例。该问题描述了两个共享固定大小的线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中,然后重复此过程。与此同时,消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据,消费者也不会在缓冲区中空时消耗数据。
要解决该问题,就必须让生产者在缓冲区满时休眠(要么干脆就放弃数据),等到下次消费者消耗缓冲区中的数据的时候,生产者才能被唤醒,开始往缓冲区添加数据。同样,也可以让消费者在缓冲区空时进入休眠,等到生产者往缓冲区添加数据之后,再唤醒消费者。通常采用的方法解决该问题,常用的方法有
生产者消费者的实现
这儿是用阻塞队列LinkedBlockingQueue来实现的,阻塞队列
package com.a.consumer; import java.util.concurrent.*; public class consumer3 { // 建立一个阻塞队列 private LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<Object>(10); public consumer3() { } public void start() { new Producer().start(); new Consumer().start(); } public static void main(String[] args) throws Exception { consumer3 s3 = new consumer3(); s3.start(); } class Producer extends Thread { public void run() { while (true) { try { Object o = new Object(); // 取出一个对象 queue.put(o); //队列满时会自动阻塞 System.out.println("Producer: " + o); } catch (InterruptedException e) { e.printStackTrace(); } } } } class Consumer extends Thread { public void run() { while (true) { try { // 取出一个对象 Object o = queue.take(); System.out.println("Consumer: " + o); } catch (InterruptedException e) { e.printStackTrace(); } } } } }下面研究下LinkedBlockingQueue的源码
首先看一下它的put方法
注意下面这句话,它会调用putLock.lockInterruptibly()这个方法,来试图获取这个putLock这个锁
public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); // Note: convention in all put/take/etc is to preset local var // holding count negative to indicate failure unless set. int c = -1; final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; putLock.lockInterruptibly(); try { /* * Note that count is used in wait guard even though it is * not protected by lock. This works because count can * only decrease at this point (all other puts are shut * out by lock), and we (or some other waiting put) are * signalled if it ever changes from * capacity. Similarly for all other uses of count in * other wait guards. */ while (count.get() == capacity) { notFull.await(); } enqueue(e); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); }
在看lockInterruptibly()方法的源码,实际是调用的sync这个同步器的acquireInterruptibly这个方法
public void lockInterruptibly() throws InterruptedException { sync.acquireInterruptibly(1); }
在看acquireInterruptibly这个方法,就是先检查当前线程中断标识位是不是true,是true时,将抛出中断异常,否则会试着去获取锁,当没有获得锁时,会执行doAcquireInterruptibly这个方法
public final void acquireInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (!tryAcquire(arg)) doAcquireInterruptibly(arg); }下面在看一下doAcquireInterruptibly这个方法,要注意shouldParkAfterFailedAcquire这个方法,就是当它为true时,会接着执行parkAndCheckInterrupt()这个方法,当它也为真时,会跳出当前循环,然后取消获取锁,并且同时抛出异常。
private void doAcquireInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.EXCLUSIVE); try { for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC return; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) break; } } catch (RuntimeException ex) { cancelAcquire(node); throw ex; } // Arrive here only if interrupted cancelAcquire(node); throw new InterruptedException(); }