美文网首页
LinkedBlockingQueue和ConcurrentLi

LinkedBlockingQueue和ConcurrentLi

作者: 骋_listen | 来源:发表于2019-10-08 21:17 被阅读0次

    队列是JAVA开发过程中一种非常常见的数据结构,尤其是需要再使用生产者-消费者的业务模型中,Queue常常作为多线程执行任务的数据交界点,从而保证生产者产生的数据能够依次被消费。

    阻塞队列的选择

    阻塞队列的实现包括ArrayBlockingQueue与LinkedBlockingQueue。相同点不做赘述,区别有以下几点:
    1.初始化时,ArrayBlockingQueue必须指定队列最大容量,LinkedBlockingQueue不强制指定,若不指定,默认Interger.Max为最大容量。
    2.ArrayBlockingQueue内部数据结构是数组:Element[],通过putIndex和takeIndex下标的循环移动控制队首和队尾;LinkedBlockingQueue内部结构是链表:Node<Element>,通过head 和 tail节点控制队首和队尾。
    3.ArrayBlockingQueue生产与消费之间共用一把锁,而LinkedBlockingQueue生产与消费时用不同的锁竞争。

    对于阻塞队列的选择,一方面考虑吞吐性能,另一方面考虑内存占用。

    我们可以看到上面说的第三点,可以确定的是,在多生产者与多消费者的情况下,LinkedBlockingQueue的吞吐性能肯定是要更高的,而且ArrayBlockingQueue在初始化时直接就申请了一片连续的内存空间。所以在实际生产使用环境中,没有特殊限制考虑,我们在使用阻塞队列时往往用LinkedBlockingQueue。

    那什么场景下我们会偏向于使用ArrayBlockingQueue呢?
    • 生产者与消费者之间没有太大竞争,倾向于单消费者,单生产者,且两者之间冲突较小,这种情况下数组寻址是明显要比链表去指向next的操作要更快的
    • 基本可以确定队列大小,且队列大小稳定在一定的数量,这个时候数组占用内存是比链表小的

    阻塞队列与非阻塞队列的选择

    首先,ConcurrentLinkedQueue相对阻塞队列来说,采用的是CAS无锁操作,没有take和put方法,主用poll与offer,无界。有人说,既然此队列内部进队和出队操作采用的是无锁,那性能肯定比有锁的BlockingQueue强,那BlockingQueue还有啥用武之地,其实不然,有些时候我们就需要线程进入阻塞状态而非不断自旋消耗CPU,我们可以归类以下场景:

    • 数据入队速度过快,出队速度过慢,这个时候ConcurrentLinkedQueue如果不借助其他限制手段,随着时间的推移,JVM必然会进行频繁的FULL GC ,严重的情况下甚至会发生OOM。使用BlockingQueue可以更好的控制内存的状况。
    • 数据入队速度过慢,出队速度过快,这个时候消费者线程如果一定想要拿到数据而不进行阻塞,将进入大量时间的自旋状态,白白浪费CPU资源。
    • 入队与出队速度相仿。
      这时候要考虑速度,有多少个线程在同时做操作,线程操作的频率如何?
      在大部分场景下,ConcurrentLinkedQueue的性能是要比BlockingQueue要好的,注意是大部分,如果线程之间的竞争足够又高又快,CAS操作的CPU消耗以及线程操作的成功率是极低的,这个时候是会反而不如用锁竞争控制效率来的高。
      我们写了个测试类可以大致看下观感下,在同样的环境下,消费者与生产者在不断对队列进行操作,然后不断增加消费者与生产者内部线程的数量。
    package algorithm;
    
    import java.util.LinkedList;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ConcurrentLinkedQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.ThreadLocalRandom;
    import java.util.concurrent.atomic.LongAdder;
    
    public class TestBlockingQueue {
    
        BlockingQueue<Data> linkedBlockingQueue = new LinkedBlockingQueue<>(100);
    //    BlockingQueue<Data> linkedBlockingQueue = new ArrayBlockingQueue<>(100);
        LongAdder longAdder = new LongAdder();
        Producer producer;
        Consumer consumer;
    
        static class Data {
            String msg;
    
            public Data(String msg) {
                this.msg = msg;
            }
        }
    
        TestBlockingQueue(int size) {
            producer = new Producer(size);
            consumer = new Consumer(size);
        }
    
        public void startTest() {
            producer.startProduce();
            consumer.startConsume();
        }
    
        public long stopTestAndReturn() {
            producer.stopProduce();
            consumer.stopConsume();
            return longAdder.longValue();
        }
    
    
        class Producer{
    
            ExecutorService service;
            List<ProduceWorker> workers = new LinkedList<>();
            Producer(int concurrentNum) {
                service = Executors.newFixedThreadPool(concurrentNum);
                for(int i = 0; i < concurrentNum; i++) {
                    workers.add(new ProduceWorker());
                }
            }
    
            public void startProduce() {
                workers.forEach(worker -> service.execute(worker));
            }
    
            public void stopProduce() {
                service.shutdownNow();
            }
    
            class ProduceWorker implements Runnable {
    
                @Override
                public void run() {
                    for(;;) {
                        if(!Thread.currentThread().isInterrupted()) {
                            try {
                                linkedBlockingQueue.put(new Data(randomString(10)));
                            } catch (InterruptedException e) {
    //                        e.printStackTrace();
                            }
                        } else {
                            break;
                        }
                    }
                }
            }
    
        }
    
        class Consumer{
            ExecutorService service;
    
            List<ConsumeWork> workers = new LinkedList<>();
    
            Consumer(int concurrentNum) {
                service = Executors.newFixedThreadPool(concurrentNum);
                for(int i = 0; i < concurrentNum; i++) {
                    workers.add(new ConsumeWork());
                }
            }
    
            class ConsumeWork implements Runnable {
    
                @Override
                public void run() {
                    for(;;) {
                        if(!Thread.currentThread().isInterrupted()) {
                            try {
                                Data data = linkedBlockingQueue.take();
                                if (null != data) {
                                    longAdder.increment();
                                }
                            } catch (InterruptedException e) {
    //                        e.printStackTrace();
                            }
                        } else {
                            break;
                        }
                    }
                }
            }
    
            public void startConsume() {
                workers.forEach(worker -> service.execute(worker));
            }
    
            public void stopConsume() {
                service.shutdownNow();
            }
        }
    
    
        public static void main(String[] args) {
            long timeStart = System.currentTimeMillis();
            System.out.println(timeStart);
            TestBlockingQueue testQueue = new TestBlockingQueue(4);
            testQueue.startTest();
            try {
                Thread.currentThread().sleep(5000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            long result = testQueue.stopTestAndReturn();
            System.out.println("最终结果为; " + result);
            long timeEnd = System.currentTimeMillis();
            System.out.println(timeEnd);
            //计算每s吞吐
            double average = (result / (timeEnd -timeStart)) * 1000;
            System.out.println("1每秒吞吐: " + average);
        }
    
        static String randomString(int strLength) {
            Random rnd = ThreadLocalRandom.current();
            StringBuilder ret = new StringBuilder();
            for (int i = 0; i < strLength; i++) {
                boolean isChar = (rnd.nextInt(2) % 2 == 0);// 输出字母还是数字
                if (isChar) { // 字符串
                    int choice = rnd.nextInt(2) % 2 == 0 ? 65 : 97; // 取得大写字母还是小写字母
                    ret.append((char) (choice + rnd.nextInt(26)));
                } else { // 数字
                    ret.append(Integer.toString(rnd.nextInt(10)));
                }
            }
            return ret.toString();
        }
    }
    

    ConcurrentQueue无法用put和take方法,需要用poll和offer,其他代码一致,不同的地方在于

                            Data data = concurrentLinkedQueue.poll();
                            if (null != data) {
                                longAdder.increment();
                            }
    
    concurrentLinkedQueue.offer(new Data(randomString(10)));
    

    可以先试试同时有4个生产者和消费者在不断进行队列操作,然后再试试1000个生产者与消费者在不断进行队列操作。

    BlockingQueue-4
    ![ConcurrentQueue-4](https://img.haomeiwen.com/i14745967/9f2a71o z66c7c3f219.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
    这个情况下可以看出无锁操作是远高于有锁操作的
    BlockingQueue-100
    ConcurrentQueue-100
    负载过高的情况下,CAS效率低下,反而不如有锁操作

    相关文章

      网友评论

        本文标题:LinkedBlockingQueue和ConcurrentLi

        本文链接:https://www.haomeiwen.com/subject/giqmuctx.html