devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

java 阻塞队列_linkedlist put方法阻塞

作者：从前慢现在也慢 | 2024-08-12 07:58:38

赞

踩

linkedlist put方法阻塞

Queue

队列是先进先出

LinkedList实现了Queue接口。Queue接口窄化了对LinkedList的方法的访问权限（即在方法中的参数类型如果是Queue时，就完全只能访问Queue接口所定义的方法 了，而不能直接访问 LinkedList的非Queue的方法），以使得只有恰当的方法才可以使用。BlockingQueue 继承了Queue接口

BlockingQueue

BlockingQueue 继承了Queue接口。定义了一种阻塞的FIFO queue，每一个BlockingQueue都有一个容量，让容量满时往BlockingQueue中添加数据时会造成阻塞，当容量为空时取元素操作会阻塞。 阻

塞队列的操作可以根据它们的响应方式分为以下三类：aad、removee和element操作在你试图为一个已满的队列增加元素或从空队列取得元素时 抛出异常。当然，在多线程程序中，队列在任何时间都可能变成满的或空的，所以你可能想使用offer、poll、peek方法。这些方法在无法完成任务时 只是给出一个出错示而不会抛出异常。

ArrayBlockingQueue

ArrayBlockingQueue是一个由数组支持的有界阻塞队列。在读写操作上都需要锁住整个容器，因此吞吐量与一般的实现是相似的，适合于实现“生产者消费者”模式。
ArrayBlockingQueue在构造时需要指定容量， 并可以选择是否需要公平性，如果公平参数被设置true，等待时间最长的线程会优先得到处理（其实就是通过将ReentrantLock设置为true来 达到这种公平性的：即等待时间最长的线程会先操作）。通常，公平性会使你在性能上付出代价，只有在的确非常需要的时候再使用它。它是基于数组的阻塞循环队 列，此队列按 FIFO（先进先出）原则对元素进行排序。

锁相关

底层锁使用 ReentrantLock 通过 Condition 队列 控制读（notEmpty）、写（notFull）。

	 /** The queued items 数组结构 */
    final Object[] items;
    
    /** Main lock guarding all access */
    final ReentrantLock lock;

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

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

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

Condition 为一个等待队列。将等待的线程放入次队列，等待获取锁

可通过参数空值获取锁策略。默认为非公平获取锁

  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();
    }


    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

LinkedBlockingQueue

LinkedBlockingQueue 基于链表的阻塞队列，同ArrayListBlockingQueue类似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于同样的原理。而LinkedBlockingQueue之所以能够高效的处理并发数据，还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。

锁相关

底层锁使用 ReentrantLock 有 读锁和写锁两个锁

     /** 链表结构 */
    transient Node<E> head;

    /**
     * Tail of linked list.
     * Invariant: last.next == null
     */
    private transient Node<E> last;

    /** Lock held by take, poll, etc */
    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();

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20

底层锁使用 ReentrantLock 分别为两个锁 控制读（takeLock）、写（putLock）。以及 Condition队列控制

SynchronousQueue：

SynchronousQueue 是一个特殊的队列，它的内部同时只能够容纳单个元素。如果该队列已有一元素的话，试图向队列中插入一个新元素的线程将会阻塞，直到另一个线程将该元素从队列中抽走。同样，如果该队列为空，试图向队列中抽取一个元素的线程将会阻塞，直到另一个线程向队列中插入了一条新的元素。据此，把这个类称作一个队列显然是夸大其词了。它更多像是一个汇合点。

/**
     * Saves this queue to a stream (that is, serializes it).
     * @param s the stream
     * @throws java.io.IOException if an I/O error occurs
     */
    private void writeObject(java.io.ObjectOutputStream s)
        throws java.io.IOException {
        boolean fair = transferer instanceof TransferQueue;
        if (fair) {
            qlock = new ReentrantLock(true);
            waitingProducers = new FifoWaitQueue();
            waitingConsumers = new FifoWaitQueue();
        }
        else {
            qlock = new ReentrantLock();
            waitingProducers = new LifoWaitQueue();
            waitingConsumers = new LifoWaitQueue();
        }
        s.defaultWriteObject();
    }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
 private static final sun.misc.Unsafe UNSAFE;
  static {
                try {
                    UNSAFE = sun.misc.Unsafe.getUnsafe();
                    Class<?> k = SNode.class;
                    matchOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("match"));
                    nextOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("next"));
                } catch (Exception e) {
                    throw new Error(e);
                }
            }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

使用了魔法类unsafe 和 ReentrantLock 锁

ArrayBlockingQueue和LinkedBlockingQueue的区别：

  1. 队列中锁的实现不同
    ArrayBlockingQueue实现的队列中的锁是没有分离的，即生产和消费用的是同一个锁；
    LinkedBlockingQueue实现的队列中的锁是分离的，即生产用的是putLock，消费是takeLock

  2. 在生产或消费时操作不同
    ArrayBlockingQueue实现的队列中在生产和消费的时候，是直接将枚举对象插入或移除的；
    LinkedBlockingQueue实现的队列中在生产和消费的时候，需要把枚举对象转换为Node进行插入或移除，会影响性能

  3. 队列大小初始化方式不同
    ArrayBlockingQueue实现的队列中必须指定队列的大小；
    LinkedBlockingQueue实现的队列中可以不指定队列的大小，但是默认是Integer.MAX_VALUE

还有带超时的offer和poll方法变种，例如，下面的调用：
boolean success = q.offer(x,100,TimeUnit.MILLISECONDS);
尝试在100毫秒内向队列尾部插入一个元素。如果成功，立即返回true；否则，当到达超时进，返回false。同样地，调用：
Object head = q.poll(100, TimeUnit.MILLISECONDS);
如果在100毫秒内成功地移除了队列头元素，则立即返回头元素；否则在到达超时时，返回null。

最后，我们有阻塞操作put和take。put方法在队列满时阻塞，take方法在队列空时阻塞。

还有带超时的offer和poll方法变种，例如，下面的调用：

    boolean success = q.offer(x,100,TimeUnit.MILLISECONDS);
    尝试在100毫秒内向队列尾部插入一个元素。
    如果成功，立即返回true；否则，当到达超时进，返回false。同样地，调用：

   Object head = q.poll(100, TimeUnit.MILLISECONDS);
   如果在100毫秒内成功地移除了队列头元素，则立即返回头元素；否则在到达超时时，返回null。

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
package com.us.queue;


import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by yangyibo on 17/2/23.
 * <p>
 * add        增加一个元索                     如果队列已满，则抛出一个IIIegaISlabEepeplian异常
 * addall      增加一个 String 集合中的所有元素
 * remove     移除并返回队列头部的元素    如果队列为空，则抛出一个NoSuchElementException异常
 * removeAll;  删除队列中参数包含的元素
 * element    返回队列头部的元素             如果队列为空，则抛出一个NoSuchElementException异常
 * offer       添加一个元素并返回true       如果队列已满，则返回false
 * poll         移除并返问队列头部的元素    如果队列为空，则返回null
 * peek       返回队列头部的元素             如果队列为空，则返回null
 * put         添加一个元素                      如果队列满，则阻塞
 * take        移除并返回队列头部的元素     如果队列为空，则阻塞
 * <p>
 * <p>
 * remove、element、offer 、poll、peek 其实是属于Queue接口。
 */
public class QueueTest {
    public static void main(String[] args) {
        List<String> list=init();
//        queueTest(list);
//        arrayBlockingQueue(list);
        linkedBlockingQueue(list);
    }

    private static List<String> init() {
        List<String> list = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            list.add("abel--" + i);
        }
        return list;
    }

    private static void println(String str) {
        System.out.println("\n" + str);
    }


    private static void queueTest(List<String> list) {
        Queue<String> queue = new LinkedList<>();
        list.stream().forEach(x -> queue.offer(x));

        println("---------------------  1  --------------------------\n");
        println("队列中的元素是:" + queue);
        //移除首元素
        println(queue.poll());
        println(queue.poll());
        println("队列中的元素是:" + queue);

        println("---------------------  2  --------------------------\n");

        queue.offer("abel--" + 100);
        //删除队列中包含的list中的元素
        queue.removeAll(list);
        println("队列中的元素是:" + queue);

        println("---------------------  3  --------------------------\n");

        //增加一个string 集合
        queue.addAll(list);
        println(queue.peek());
        println("队列中的元素是:" + queue);

    }

    /**
     * 阻塞 queue
     * 阻塞操作put和take。put方法在队列满时阻塞，take方法在队列空时阻塞。
     * ArrayBlockingQueue在构造时需要指定容量
     *
     * @param list
     */
    private static void arrayBlockingQueue(List<String> list) {
        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(10);
        queue.addAll(list);
        println("---------------------  1  --------------------------\n");
        println("队列中的元素是:" + queue);

        println("---------------------  2  --------------------------\n");
        queue.remove("abel--0");
        println("队列中的元素是:" + queue);

        println("---------------------  3  --------------------------\n");


    }

    /**
     * 阻塞 queue
     * 阻塞操作put和take。put方法在队列满时阻塞，take方法在队列空时阻塞。
     * LinkedBlockingQueue在构造时需要指定容量 ,不指定容量时默认 Integer.MAX_VALUE
     */
    private static void linkedBlockingQueue(List<String> list) {
        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(9);
        list.stream().forEach(x -> {
            try {
                queue.put(x);
                //存到第10 个元素时阻塞
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        println("---------------------  1  --------------------------\n");
        println("队列中的元素是:" + queue);

    }

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115

#阻塞队列
//当队列满了再put 元素 就阻塞，当队列为空时 take 数据也阻塞数据

package com.us.queue;


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * Created by yangyibo on 17/2/23.
 */
public class ArrayBlockingQueueTest {
    private static ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(5);

    public static void main(String[] args) {
        Thread readThread = new Thread(new ReadThread());
        readThread.start();
        for (int i = 0; i < 3; i++) {
            try {
                Thread writeThread = new Thread(new WriteThread());
                writeThread.start();
                Thread.sleep(20000);
            } catch (Exception e) {

            }
        }
    }

    private static List<String> init() {
        List<String> list = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            list.add("abel--" + i);
        }
        return list;
    }

    private static void println(String str) {
        System.out.println(str);
    }


    static class WriteThread extends Thread {

        @Override
        public void run() {
            init().stream().forEach(x -> {
                try {
                    println(x + "-------放入队列");
                    //存数据到队列当队列满了就阻塞,当数据被取走后,继续往队列中方数据
                    queue.put(x);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }


    static class ReadThread extends Thread {
        @Override
        public void run() {
            try {
                while (true) {
                    //不停的取出数据,当队列为空的时候阻塞,等待数据放入队列
                    Thread.sleep(1000);
                    println(queue.take() + "-----------取出队列");
                    //队列为空的时候阻塞
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73

参考资料：http://www.cnblogs.com/end/archive/2012/10/25/2738493.html
http://chenjumin.iteye.com/blog/2182322
undefined
undefined

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号