阻塞队列_Java 开发基础与工程实践

任何有效的生产者-消费者问题解决方案都是通过控制生产者 put()方法（生产资源）和消费者 take()方法（消费资源）的调用来实现的，一旦你实现了对方法的阻塞控制，那么你将解决该问题。Java 通过 BlockingQueue 提供了开箱即用的支持来控制这些方法的调用（一个线程创建资源，另一个消费资源）。java.util.concurrent 包下的 BlockingQueue 接口是一个线程安全的可用于存取对象的队列。

Blocking Queue 示意

BlockingQueue 是一种数据结构，支持一个线程往里存资源，另一个线程从里取资源。这正是解决生产者消费者问题所需要的，那么让我们开始解决该问题吧。

import java.util.concurrent.BlockingQueue;
class Producer implements Runnable {
    protected BlockingQueue<Object> queue;
    Producer(BlockingQueue<Object> theQueue) {
        this.queue = theQueue;
    }
    public void run() {
        try {
            while (true) {
                Object justProduced = getResource();
                queue.put(justProduced);
                System.out.println("Produced resource - Queue size now = "
                        + queue.size());
            }
        } catch (InterruptedException ex) {
            System.out.println("Producer INTERRUPTED");
        }
    }
    Object getResource() {
        try {
            Thread.sleep(100); // simulate time passing during read
        } catch (InterruptedException ex) {
            System.out.println("Producer Read INTERRUPTED");
        }
        return new Object();
    }
}Copy to clipboardErrorCopied

这里，生产者线程创建一个资源（Object 对象）并将它存入队列中，如果队列已经满了（最大为 20），它将会等待直到消费者线程从队列中取走资源，所以队列的长度永远不会超过最大值，比如 20。

class Consumer implements Runnable {
    protected BlockingQueue<Object> queue;
    Consumer(BlockingQueue<Object> theQueue) {
        this.queue = theQueue;
    }
    public void run() {
        try {
            while (true) {
                Object obj = queue.take();
                System.out.println("Consumed resource - Queue size now = "
                        + queue.size());
                take(obj);
            }
        } catch (InterruptedException ex) {
            System.out.println("CONSUMER INTERRUPTED");
        }
    }
    void take(Object obj) {
        try {
            Thread.sleep(100); // simulate time passing
        } catch (InterruptedException ex) {
            System.out.println("Consumer Read INTERRUPTED");
        }
        System.out.println("Consuming object " + obj);
    }
}Copy to clipboardErrorCopied

队列中有资源的话，消费者线程将从队列中取走资源，否则会等待直到生产者创建资源。

public class ProducerConsumerExample {
    public static void main(String[] args) throws InterruptedException {
        int numProducers = 4;
        int numConsumers = 3;
        BlockingQueue<Object> myQueue = new LinkedBlockingQueue<>(20);
        for (int i = 0; i < numProducers; i++) {
            new Thread(new Producer(myQueue)).start();
        }
        for (int i = 0; i < numConsumers; i++) {
            new Thread(new Consumer(myQueue)).start();
        }
        // Let the simulation run for, say, 10 seconds
        Thread.sleep(10 * 1000);
        // End of simulation - shut down gracefully
        System.exit(0);
    }
}
当我们运行以上代码，会发现类似如下输出结果：
Consumed resource - Queue size now = 1
Produced resource - Queue size now = 1
Consumed resource - Queue size now = 1
Consumed resource - Queue size now = 1
Produced resource - Queue size now = 1
Produced resource - Queue size now = 1
Produced resource - Queue size now = 1
Consuming object java.lang.Object@14c7f728
Consumed resource - Queue size now = 0
Consuming object java.lang.Object@2b71e323
Consumed resource - Queue size now = 0
Produced resource - Queue size now = 0
Produced resource - Queue size now = 1
Produced resource - Queue size now = 2
Consuming object java.lang.Object@206dc00b
Consumed resource - Queue size now = 1
Produced resource - Queue size now = 2
Produced resource - Queue size now = 3
Consuming object java.lang.Object@1a000bc0
Consumed resource - Queue size now = 2
Consuming object java.lang.Object@25b6183d
Consumed resource - Queue size now = 1
Produced resource - Queue size now = 2
Produced resource - Queue size now = 3
...
...
Produced resource - Queue size now = 20
Consuming object java.lang.Object@2b3cd3a6
Consumed resource - Queue size now = 19
Produced resource - Queue size now = 20
Consuming object java.lang.Object@3876982d
Consumed resource - Queue size now = 19
Produced resource - Queue size now = 20

下一节：今天在学习CAS的时候，想手写一个CAS的计数器，与JDK中的Atomic(AtomicLong，AtomicInteger等)系列的做个比较，本想性能应该能比JDK要差一丢丢，但却加了一个让线程让出时间片的代码，性能反而更高。