首页 > 代码库 > java消费者生产者模式及JDK之阻塞队列LinkedBlockingQueue实现

java消费者生产者模式及JDK之阻塞队列LinkedBlockingQueue实现

2024-08-10 04:39:51   210人阅读

生产者消费者问题

   （英语：Producer-consumer problem），也称有限缓冲问题（英语：Bounded-buffer problem），是一个多线程同步问题的经典案例。该问题描述了两个共享固定大小缓冲区的线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。

    要解决该问题，就必须让生产者在缓冲区满时休眠（要么干脆就放弃数据），等到下次消费者消耗缓冲区中的数据的时候，生产者才能被唤醒，开始往缓冲区添加数据。同样，也可以让消费者在缓冲区空时进入休眠，等到生产者往缓冲区添加数据之后，再唤醒消费者。通常采用进程间通信的方法解决该问题，常用的方法有信号灯法[1]等。如果解决方法不够完善，则容易出现死锁的情况。出现死锁时，两个线程都会陷入休眠，等待对方唤醒自己。该问题也能被推广到多个生产者和消费者的情形。

现实中的应用

     比如一个饭店，它有一个厨师和一个服务员。这个服务员必须等待厨师准备好食物。当厨师准备好时，他会通知服务员，之后服务员上菜，然后返回继续等待。这是一个任务协作的实例：厨师代表生产者，服务员代表消费者。两个任务必须在食物被生产和消费时进行握手，而系统必须以有序的方式关闭。

      可以用下图来表示这种关系。

     

技术分享


生产者消费者的实现

这儿是用阻塞队列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();
    }




java消费者生产者模式及JDK之阻塞队列LinkedBlockingQueue实现


联系
我们