Pratice Of Concurrency(一: BlockingQueue )

Java java.util.currency 包中的 BlockingQueue 接口意味着线程可以在一个队列中添加和提取对象.在接下来的段落中,我们将学习如何使用 BlockingQueue 接口.

BlockingQueue的使用

BlockingQueue典型应用是用于一个线程生产对象,另一个线程消费对象.

下面是一个原理描述图:

一个阻塞队列伴随着一个线程往队列中添加对象,另一个线程从队列中提取对象.

生产线程负责生产新对象并发它添加进队列中,直到达到队列所能容纳的边界值为止.即达到容量限制.如果队列达到容量限制,则生产线程在插入新对象时将会发生阻塞.直到有一个消费线程从队列中取走一个对象为止.

消费线程负责从阻塞队列中取出对象并处理它们.如果消费线程尝试从空队列中取出空队列,那么消费线程将会阻塞直到生产线程往队列中放入一个对象为止.

BlockingQueue方法

一个阻塞队列包含四种截然不同的方法集,用于完成在队列中插入,移除和检查元素.每种集合的不同之处在于操作请求不能被立即执行时的不同行为.下面是这些方法的列表:

四种不同的行为结合如下:

1. Throws Exception: 当尝试操作不能被立即执行时, 将抛出一个异常
2. Special Value: 当尝试操作不能被立即执行时, 指定一个返回值(通常为 true / false)
3. Blocks: 当尝试操作不能被立即执行时, 阻塞方法调用直到能够执行成功为止
4. Time Out: 当尝试操作不能被立即执行时, 阻塞方法调用直到能够执行成功为止, 但等待不能超过指定的超时时长.若超出时长, 无论成功或失败返回一个指定的值.(典型的例子是 true / false)

不能插入 null 到BlockingQueue中.如果你这么做了,BlockingQueue将抛出一个 NullPointerException .

当然我们也能够访问BlockingQueue的所有元素, 不只是队头和队尾. 例如, 你需要出队一个对象用于执行,但你的应用决定取消它.那么你仍然可以使用类似remove(o)方法来移除队列中指定的对象.然而这不是很高效,所以你不应该使用这类方法, 除非你真的需要这么做.

BlockingQueue的多种实现

BlockingQueue只是一个接口, 所以你需要使用它众多实现的其中一种来使用它. java.util.concurrency 包中有如下几种BlockingQueue的接口实现(在 Java 6中):

• ArrayBlockingQueue
• DelayQueue
• LinkedBlockingQueue
• PriorityBlockingQueue
• SynchronousQueue

Java BlockingQueue实例

下面是一个Java BlockingQueue的实例. 实例中使用BlockingQueue接口的实现类,ArrayBlockingQueue.

首先,BlockingQueueExample类中分别启动一个生产和消费线程.生产线程负责插入字符串到共享BlockingQueue中,而消费线程则负责取走它们.

public class ArrayBlockingQueueExample {
    public static void main(String[] args) {
        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(1024);
        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);

        new Thread(producer).start();
        new Thread(consumer).start();

        try {
            Thread.sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
复制代码

下面是Producer类.注意我们每调用一次put()就会让线程睡眠一秒.这会导致消费线程在队列中等待对象时发生阻塞.

public class Producer implements Runnable {
    private ArrayBlockingQueue<String> queue;

    public Producer(ArrayBlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            queue.put("1");
            Thread.sleep(1000);
            queue.put("2");
            Thread.sleep(1000);
            queue.put("3");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
复制代码

下面是Consumer类.它只是负责从队列中取出字符串并通过System.out打印出来.

package org.menfre;

import java.util.concurrent.ArrayBlockingQueue;

public class Consumer implements  Runnable {
    private ArrayBlockingQueue<String> queue;

    public Consumer(ArrayBlockingQueue<String> queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            System.out.println(queue.take());
            System.out.println(queue.take());
            System.out.println(queue.take());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
复制代码

该系列博文为笔者复习基础所著译文或理解后的产物,复习原文来自Jakob Jenkov所著Java.util.concurrent

Next: ArrayBlockingQueue