并发编程(10)-queue

冷不防 2022-06-06 09:28 153阅读 0赞

# 1.ConcurrentLinkedQueue #

concurrentLinkedQueue是一个适用于高并发场景下的队列，通过无锁的方式，实现高并发状态下的高性能，通常concurrentLinkedQueue性能好于BlockingQueue。它是一个基于链接节点的无界线程安全队列。该队列的元素遵循先进先出的原则。头是最先加入的，尾是最近加入的，该队列不允许null元素。  
concurrentLinkedQueue重要方法：  
add()和offer()都是加入元素的方法（concurrentLinkedQueue中，这两个方法没有任何区别）；  
poll()和peak()都是取头元素节点，区别在于前者会删除元素，后者不会。

public class UseConcurrentLinkedQueue {
    
        public static void main(String[] args) {
            ConcurrentLinkedDeque<String> cld = new ConcurrentLinkedDeque<String>();
            cld.offer("a");
            cld.offer("b");
            cld.offer("c");
            cld.offer("d");
            cld.add("e");
    
            System.out.println(cld.poll());
            System.out.println(cld.size());
            System.out.println(cld.peek());
            System.out.println(cld.size());
        }
    }

打印结果：

> a  
> 4  
> b  
> 4

# 2.BlockingQueue #

**ArrayBlockingQueue**：基于数组的阻塞队列实现，在ArrayBlockingQueue内部，维护了一个定长数组，以便缓存队列中的数据对象，其内部实现读写分离，也就意味着生产和消费不能完全并行，长度是需要定义的，可以指定先进先出或者先进后出，也叫**有界队列**，在很多场合非常适合使用。

public static void main(String[] args) throws Exception {
            //创建一个有界队列，长度为5
            ArrayBlockingQueue<String> abq = new ArrayBlockingQueue<String>(5);
            abq.put("a");
            abq.add("b");
            abq.add("c");
            abq.add("d");
            abq.add("e");
            //3秒钟无法加入到队列中返回false，加入进去返回true
            System.out.println(abq.offer("f", 3, TimeUnit.SECONDS));
        }

打印结果：

> false因为队列的长度为5

如果在打印之前再增加一个元素，结果会直接报错，显示队列已经满了。

public static void main(String[] args) throws Exception {
            //创建一个有界队列，长度为5
            ArrayBlockingQueue<String> abq = new ArrayBlockingQueue<String>(5);
            abq.put("a");
            abq.add("b");
            abq.add("c");
            abq.add("d");
            abq.add("e");
            abq.add("f");//新增加一个
            //3秒钟无法加入到队列中返回false，加入进去返回true
            System.out.println(abq.offer("f", 3, TimeUnit.SECONDS));
        }

打印结果：

> Exception in thread “main” java.lang.IllegalStateException: Queue full  
> at java.util.AbstractQueue.add(AbstractQueue.java:98)  
> at java.util.concurrent.ArrayBlockingQueue.add(ArrayBlockingQueue.java:312)  
> at com.thread.mythread.conn014.UseBlockingQueue.main(UseBlockingQueue.java:18)

**LinkedBlockingQueue**：基于链表的阻塞队列，通ArrayBlockingQueue类似，其内部也维持着一个暑假缓冲队列（该队列由一个链表构成），LinkedBlockingQueue之所以能够高效的处理并发数据，是因为其内部实现采用分离锁（读写分离两个锁），从而实现生产者和消费者操作的完全并行运行，是一个**无界队列**。

public static void main(String[] args) {
            LinkedBlockingDeque<String> lbd = new LinkedBlockingDeque<String>(5);
            lbd.offer("a");
            lbd.offer("b");
            lbd.offer("c");
            lbd.offer("d");
            lbd.offer("e");
            lbd.offer("f");
            System.out.println(lbd.size());
            for (String lbdi: lbd) {
                System.out.println(lbdi);
            }
        }

打印结果：

> 5  
> a  
> b  
> c  
> d  
> e

这里定义的队列的长度为5（也可以不指定长度，则为无界），但是在执行的过程中，虽然添加了六个元素却并没有报错。但是第六个元素不能使用add()方法。

**SynchronousQueue**：一种没有缓冲的队列，生产者产生的数据直接会被消费者获取并消费。

public static void main(String[] args) {
        SynchronousQueue<String> sq = new SynchronousQueue<String>();
            sq.add("a");
        }

打印结果：不能添加元素

> Exception in thread “main” java.lang.IllegalStateException: Queue full  
> at java.util.AbstractQueue.add(AbstractQueue.java:98)  
> at com.thread.mythread.conn014.UseBlockingQueue.main(UseBlockingQueue.java:41)

**PriorityBlockingQueue**：基于优先级的阻塞队列（优先级的判断通过构造函数传入的Compator对象来决定，也就是说传入队列的对象必须实现Comparable接口），在实现PriorityBlockingQueue时，内部控制线程同步的锁，采用的是公平锁，他也是一个无界的队列。  
**DelayQueue**：带有延迟时间的Queue，其中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue中的元素必须实现Delayed接口，DelayQueue是一个没有大小限制的队列，应用场景很多，比如对缓存超时的数据进行移除、任务超时处理、空闲连接的关闭等等。