java Blocking Queue

一、Java中的阻塞队列

　　在多线程之间通信中，多个线程共享一个进程所分配的资源，共享内存是一种常见的通信方式，而阻塞队列则是其实现方式的一种，例如经典的生产者-消费者模式。

　　A Queue that addtionally supports operations that wait for the queue to become non-empty when retrieving an element, and  wait for space to become avialable in the queue when storing an element.

　　阻塞队列中提供了2个操作：

　　　　队列为空时，获取元素的线程会阻塞队列一直至队列非空。

　　　　队列为满时，存储元素的线程会阻塞队列非满。

　　

　　

　　Java中的阻塞队列有7种：

  • ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列。
  • LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列。
  • PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列。
  • DelayQueue：一个使用优先级队列实现的无界阻塞队列。
  • SynchronousQueue：一个不存储元素的阻塞队列。
  • LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
  • LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

　　下面进行说明

二、ArrayBlockingQueue

使用数组实现队列

2.1 构造器

public ArrayBlockingQueue(int capacity,
                          boolean fair,
                          @NotNull Collection<? extends E> c)
//Creates an ArrayBlockingQueue with the given (fixed) capacity, the specified access policy and initially containing the elements of the given collection, added in traversal order of the collection‘s iterator.

参数说明：

　　capacity: 队列的容量

　　fair: 默认是false，如果是true的话则移除元素的顺序符合FIFO顺序，false的话没有顺序

　　c: 预填充元素

实现主要如下：

 public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

2.2 put方法实现

/** Main lock guarding all access */
    final ReentrantLock lock;

    /** Condition for waiting takes */
    private final Condition notEmpty;

    /** Condition for waiting puts */
    private final Condition notFull;

由于数组这种数据结构的特殊性，若想要线程安全的添加或者删除，都必须将整个数组锁住，因此这里实现使用一把锁, 以及2个条件队列。

public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

这段代码非常容易理解，注意点：

(1) 可重入锁

(2) 使用2个条件队列:Java concurrent包对多条件队列的支持，古老的wait和notify方法也可以实现，只是由于条件队列中的条件不同，必须使用notifyall()，损失了性能。

(3) 可中断锁

2.3 take() 方法说明

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

总结： ArrayBlockingQueue应该是最简单的阻塞队列实现了，由于数组结构的特殊性，使用了一把锁和2个条件队列，锁的方式是可中断锁。

三、LinkedBlockingQueue

使用链表实现队列，构造器使用方式和ArrayBlockingQueue一样

 public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }

注意到初始化的时候创建了一个空节点...

3.1 使用2把锁实现

这里使用了2把锁，2个条件队列实现，而且在属性定义中就直接初始化了。一个存锁，一个取锁，一个非空条件队列，一个非满条件队列

　　private final ReentrantLock takeLock = new ReentrantLock();

    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();

    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();

    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

3.2 put的实现

 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;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

说明：

  • 除了锁之外，使用了原子变量来记录链表大小，因为使用了2把锁来锁节点，全局链表的大小使用线程安全的原子变量确实比较合适
  • 此时锁的是putLock，存锁

下面看singalNotEmpty()方法:

 private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

发现唤醒时使用了取锁，使用了连续的锁。我们大致梳理一下顺序，在不考虑异常的情况下:

(1) 当前线程竞争存锁putLock并将其锁住

(2) 条件队列notFull(非满条件)等待

(3) 链表中插入元素

(4) 原子变量自增

(5) if(c+1<capacity)，则条件队列notFull唤醒操作，通知其他的put线程，链表未达上限，依旧可以插入元素

(6) 释放putLock锁

(7) 如果c==0，原子变量修改-1->0，表明插入成功，通知其他take线程可以去取出元素

　　i. 当前线程竞争takeLock，并且锁住

　　ii. notEmpty.signal，通知其他在wait take lock的线程可以取出元素

　　iii. 释放takeLock

3.3 take的实现

类似，略

3.4 总结

(1) 使用了2把不同的锁

(2) 使用原子变量控制容量

(3) 使用链表数据结构

为什么可以用2把锁提升性能，减少锁竞争？
注意下面2个方法：入队方法操作的是last，出对方法操作的first，正是由于链表结构的特殊性，可以使用2把锁来提高锁粒度，从而减少锁竞争。

 private void enqueue(Node<E> node) {
        // assert putLock.isHeldByCurrentThread();
        // assert last.next == null;
        last = last.next = node;
    }

    /**
     * Removes a node from head of queue.
     *
     * @return the node
     */
    private E dequeue() {
        // assert takeLock.isHeldByCurrentThread();
        // assert head.item == null;
        Node<E> h = head;
        Node<E> first = h.next;
        h.next = h; // help GC
        head = first;
        E x = first.item;
        first.item = null;
        return x;
    }

四、Synchronousqueue

个人感觉http://ifeve.com/java-synchronousqueue/讲的比较详细.这里引用一段话描述：

不像ArrayBlockingQueue或LinkedListBlockingQueue，SynchronousQueue内部并没有数据缓存空间，你不能调用peek()方法来看队列中是否有数据元素，因为数据元素只有当你试着取走的时候才可能存在，不取走而只想偷窥一下是不行的，当然遍历这个队列的操作也是不允许的。队列头元素是第一个排队要插入数据的线程，而不是要交换的数据。数据是在配对的生产者和消费者线程之间直接传递的，并不会将数据缓冲数据到队列中。可以这样来理解：生产者和消费者互相等待对方，握手，然后一起离开。

package concurrent.demo.synchronous_queue;

import concurrent.demo.Utils;

import java.util.concurrent.CountDownLatch;

/**
 * <B>系统名称：</B><BR>
 * <B>模块名称：</B><BR>
 * <B>中文类名：</B><BR>
 * <B>概要说明：</B><BR>
 *
 * @author carl.yu
 * @since 2016/6/14
 */
public class NativeSynchronousQueue<E> {

    E item = null;
    boolean putting = false;

    //保证放进去之后，其他的取走了，其他的put线程才允许取
    public synchronized void put(E e) throws InterruptedException {
        //1. 只有一个item为null的时候才允许放
        while (putting) {
            wait();
        }
        putting = true;
        item = e;
        //2. 通知其他的线程来取，可能会通知错了人，造成假唤醒，所以需要用条件队列putting为true来判定，继续睡吧
        notifyAll(); /*必须使用notify不能使用notifyAll*/
        /*这里的notifyAll是为了让其他的take线程醒来，而不是put线程醒来哦*/

        //3. 只有取完的线程才可以来拿
        while (item != null) {
            wait();
        }
        putting = false;
        notifyAll();
        /*这里才是为了put线程醒来*/
    }

    /*take比较简单，就是拿*/
    public synchronized E take() throws InterruptedException {
        E res = null;
        while (item == null) {
            wait();
        }
        res = item;
        item = null;
        notifyAll();
        return res;
    }

    //测试
    public static void main(String[] args) throws Exception {
        final NativeSynchronousQueue queue = new NativeSynchronousQueue();
        final CountDownLatch latch = new CountDownLatch(3);
        Thread put01 = new Thread() {
            @Override
            public void run() {
                try {
                    queue.put("1");
                    System.out.println("put01成功");
                    latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        Thread put02 = new Thread() {
            @Override
            public void run() {
                try {
                    queue.put("2");
                    System.out.println("put02成功");
                    latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        Thread take01 = new Thread() {
            @Override
            public void run() {
                try {
                    System.out.println("take成功:" + queue.take());
                    latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        Thread take02 = new Thread() {
            @Override
            public void run() {
                try {
                    System.out.println("take成功:" + queue.take());
                    latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        put01.start();
        Utils.sleep(10);
        put02.start();
        Utils.sleep(10);
        //当没有线程take的时候，put永远不会成功
        take01.start();
        Utils.sleep(10);
        take02.start();
        latch.await();
        System.out.println("测试结束");
    }

五、条件队列和线程池的使用

Java中，线程池的使用非常常见，jdk1.8中还新增了许多连接池，例如newWorkStealingPool用作并行计算...

这里主要强调条件队列的用法

5.1 使用有界队列实现

直接演示

定义任务类：

public class MyTask implements Runnable {

    private int taskId;
    private String taskName;

    public MyTask(int taskId, String taskName){
        this.taskId = taskId;
        this.taskName = taskName;
    }

    public int getTaskId() {
        return taskId;
    }

    public void setTaskId(int taskId) {
        this.taskId = taskId;
    }

    public String getTaskName() {
        return taskName;
    }

    public void setTaskName(String taskName) {
        this.taskName = taskName;
    }

    public void run() {
        try {
            System.out.println("run taskId =" + this.taskId);
            Thread.sleep(5*1000);
            //System.out.println("end taskId =" + this.taskId);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public String toString(){
        return Integer.toString(this.taskId);
    }

}

自定义拒绝策略:

package com.bjsxt.height.concurrent018;

import java.net.HttpURLConnection;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class MyRejected implements RejectedExecutionHandler{

    public MyRejected(){
    }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("自定义处理..");
        System.out.println("当前被拒绝任务为：" + r.toString());

    }

}
public static void main(String[] args) {
        /**
         * 在使用有界队列时，若有新的任务需要执行，如果线程池实际线程数小于corePoolSize，则优先创建线程，
         * 若大于corePoolSize，则会将任务加入队列，
         * 若队列已满，则在总线程数不大于maximumPoolSize的前提下，创建新的线程，
         * 若线程数大于maximumPoolSize，则执行拒绝策略。或其他自定义方式。
         *
         */
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                1,                 //coreSize
                2,                 //MaxSize
                60,             //60
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(3)            //指定一种队列 （有界队列）
                //new LinkedBlockingQueue<Runnable>()
                , new MyRejected()
                //, new DiscardOldestPolicy()
                );

        MyTask mt1 = new MyTask(1, "任务1");
        MyTask mt2 = new MyTask(2, "任务2");
        MyTask mt3 = new MyTask(3, "任务3");
        MyTask mt4 = new MyTask(4, "任务4");
        MyTask mt5 = new MyTask(5, "任务5");
        MyTask mt6 = new MyTask(6, "任务6");

        pool.execute(mt1);
        pool.execute(mt2);
        pool.execute(mt3);
        pool.execute(mt4);
        pool.execute(mt5);
        pool.execute(mt6);

        pool.shutdown();

    }

逐渐增加任务数，由1增加到6发现效果如下:

(1) 当任务数<1时，加入一个任务直接创建线程去处理

(2) 继续加入任务>corePoolSize=1时，会加入队列ArrayBlockingQueue

(3) 一直加入到任务数为4，队列中3个元素，队列满了

(4) 任务数继续增加，到5，会继续创建线程一直到maxPoolSize(2)，因此此时执行的任务都1和5

(5) 任务数增加为6，执行拒绝策略

5.2 无界队列实现

将上述代码修改为用LinkedBlockingQueue()无界队列实现，发现此时maxSize参数没有作用，拒绝策略也没有作用，只有coreSize才有作用

(1) 当任务数<coreSize,会创建线程

(2) 当任务数>coreSize，会将任务放置到无界队列中，直到系统崩溃

(3) maxSize没有任何作用

(4) 拒绝策略没有任何作用

因此，JDK在实现FixedThreadPool时,maxSize和coreSize相等

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

5.3 Synchronousqueue实现

JDK中CachedThreadPool用这种队列实现，性能非常好

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

　　

时间： 2024-10-13 04:36:25

java Blocking Queue的相关文章

[Java] LinkedList / Queue - 源代码学习笔记

简单地画了下 LinkedList 的继承关系,如下图.只是画了关注的部分,并不是完整的关系图.本博文涉及的是 Queue, Deque, LinkedList 的源代码阅读笔记.关于 List 接口的笔记,可以参考上一篇博文 List / ArrayList - 源代码学习笔记 Queue 1. 继承 Collection 接口,并提供了额外的插入.提取和查看元素的方法.新增的方法都有两种形式:当操作失败时,抛出异常或者返回一个特殊值.特殊值可以是 null 或者 false ,这取决于方法本

LeetCode 1188. Design Bounded Blocking Queue

原题链接在这里:https://leetcode.com/problems/design-bounded-blocking-queue/ 题目: Implement a thread safe bounded blocking queue that has the following methods: BoundedBlockingQueue(int capacity) The constructor initializes the queue with a maximum capacity.

java.util.Queue用法

Queue接口与List.Set同一级别,都是继承了Collection接口.LinkedList实现了Queue接 口.在队列这种数据结构中,最先插入的元素将是最先被删除的元素:反之最后插入的元素将是最后被删除的元素,因此队列又称为“先进先出”(FIFO—first in first out)的线性表.在java5中新增加了java.util.Queue接口,用以支持队列的常见操作.该接口扩展了java.util.Collection接口. Queue使用时要尽量避免Collection的ad

java中Queue简介

Queue: 基本上,一个队列就是一个先入先出(FIFO)的数据结构 offer,add区别:一些队列有大小限制,因此如果想在一个满的队列中加入一个新项,多出的项就会被拒绝.这时新的 offer 方法就可以起作用了.它不是对调用 add() 方法抛出一个 unchecked 异常,而只是得到由 offer() 返回的 false. poll,remove区别:remove() 和 poll() 方法都是从队列中删除第一个元素(head).remove() 的行为与 Collection 接口的版

java中Queue接口

Queue接口与List.Set同一级别,都是继承了Collection接口.LinkedList实现了Queue接 口.Queue接口窄化了对LinkedList的方法的访问权限(即在方法中的参数类型如果是Queue时,就完全只能访问Queue接口所定义的方法 了,而不能直接访问 LinkedList的非Queue的方法),以使得只有恰当的方法才可以使用.BlockingQueue 继承了Queue接口. 队列是一种特殊的线性表,它只允许在表的前端(front)进行删除操作,而在表的后端(re

使用goroutine+channel和java多线程+queue队列的方式开发各有什么优缺点？

我感觉很多项目使用java或者c的多线程库+线程安全的queue数据结构基本上可以实现goroutine+channel开发能达到的需求,所以请问一下为什么说golang更适合并发服务端的开发呢?使用goroutine+channel和java多线程+queue队列的方式开发各有什么优缺点? 使用goroutine+channel和java多线程+queue队列的方式开发各有什么优缺点? >> golang 这个答案描述的挺清楚的:http://www.goodpm.net/postreply

java集合--Queue用法

队列是一种特殊的线性表,它只允许在表的前端(front)进行删除操作,而在表的后端(rear)进行插入操作.进行插入操作的端称为队尾,进行删除操作的端称为队头.队列中没有元素时,称为空队列. 在队列这种数据结构中,最先插入的元素将是最先被删除的元素:反之最后插入的元素将是最后被删除的元素,因此队列又称为“先进先出”(FIFO—first in first out)的线性表. 在java5中新增加了java.util.Queue接口,用以支持队列的常见操作.该接口扩展了java.util.Coll

java中使用FIFO队列:java.util.Queue实现多台服务器发邮件的代码

代码下载地址:http://www.zuidaima.com/share/1838230785625088.htm 原文:java中使用FIFO队列:java.util.Queue实现多台服务器发邮件的代码 最近由于zuidaima.com注册用户的增多,qq企业邮箱发送邮件会被封禁账号导致注册后面的用户收不到周总结,所以紧急开发了一套多账号,多服务器发送邮件的程序. 大概的设计思路如下: 1.服务器可以无限扩展,但由于qq企业邮箱是限定域名,所以要想多服务器还得有多域名,多账号也不行. 2.最

Java队列Queue、双端队列Deque

http://uule.iteye.com/blog/2095650?utm_source=tuicool 注意:这都只是接口而已 1.Queue API 在java5中新增加了java.util.Queue接口,用以支持队列的常见操作.该接口扩展了java.util.Collection接口. Java代码   public interface Queue<E> extends Collection<E> 除了基本的 Collection 操作外,队列还提供其他的插入.提取和检查