Java深入学习（2）：并发队列

并发队列：

在并发队列中，jdk有两套实现：

concurrentlinkedqueue：非阻塞式队列

blockingqueue：阻塞式队列

 

阻塞式队列非阻塞式队列的区别：

阻塞式队列入列操作的时候，如果超出队列总数，这个时候会进行等待；在出列的时候，如果队列为空，也会等待

非阻塞无论如何都不等待

非阻塞效率更高，但是阻塞使用更广泛

阻塞队列的优点是能够防止队列容器溢出，防止丢失

 

非阻塞队列：

public class queuedemo {
    public static void main(string[] args) {
        concurrentlinkedqueue<string> concurrentlinkedqueue = new concurrentlinkedqueue<>();
        concurrentlinkedqueue.offer("张三");
        concurrentlinkedqueue.offer("李四");
        concurrentlinkedqueue.offer("王五");
        for (int i = 0; i < 4; i++) {
            system.out.println(concurrentlinkedqueue.poll());
        }
    }
}

打印如下：

张三
李四
王五
null

 

阻塞队列（重要）：需要初始化队列总数

public class queuedemo {
    public static void main(string[] args) throws interruptedexception {
        blockingqueue<string> arrayblockingqueue = new arrayblockingqueue<>(3);
        //添加非阻塞式队列
        arrayblockingqueue.offer("张三");
        arrayblockingqueue.offer("李四");
        arrayblockingqueue.offer("王五");
        //添加阻塞式队列，等待时间为3s
        arrayblockingqueue.offer("赵六",3, timeunit.seconds);
        system.out.println(arrayblockingqueue.poll());
        system.out.println(arrayblockingqueue.poll());
        system.out.println(arrayblockingqueue.poll());
        system.out.println(arrayblockingqueue.poll(3,timeunit.seconds));
    }
}

这种情况，等待3秒后打印：张三，李四，王五，再等待3秒后打印：null

 

换一下代码：

public class queuedemo {
    public static void main(string[] args) throws interruptedexception {
        blockingqueue<string> arrayblockingqueue = new arrayblockingqueue<>(3);
        //添加非阻塞式队列
        arrayblockingqueue.offer("张三");
        arrayblockingqueue.offer("李四");
        system.out.println(arrayblockingqueue.poll());
        arrayblockingqueue.offer("王五");
        //添加阻塞式队列，等待时间为3s
        arrayblockingqueue.offer("赵六",3, timeunit.seconds);
        system.out.println(arrayblockingqueue.poll());
        system.out.println(arrayblockingqueue.poll());
        system.out.println(arrayblockingqueue.poll());
        system.out.println(arrayblockingqueue.poll(3,timeunit.seconds));
    }
}

这种情况，立即打印张三，李四，王五，赵六，等待3秒后打印null

 

示例：

public class queuedemo {
    public static void main(string[] args) throws interruptedexception {
        blockingqueue<string> arrayblockingqueue = new arrayblockingqueue<>(3);
        //添加非阻塞式队列
        boolean success1 = arrayblockingqueue.offer("张三");
        boolean success2 = arrayblockingqueue.offer("李四");
        boolean success3 = arrayblockingqueue.offer("王五");
        //添加阻塞式队列，等待时间为3s
        boolean success4 = arrayblockingqueue.offer("赵六",3, timeunit.seconds);
        system.out.println(success1);
        system.out.println(success2);
        system.out.println(success3);
        system.out.println(success4);
    }
}

等待3秒后打印：true，true，true，false；说明赵六没有入列成功

 

生产者消费者示例：

下面模拟一个生产者消费者的例子，以便于更好地理解：

生产者线程存一个队列，消费者线程取一个队列，多线程中可以采用等待唤醒机制，在这里采用并发队列实现

package org.dreamtech;

import java.util.concurrent.blockingqueue;
import java.util.concurrent.linkedblockingqueue;
import java.util.concurrent.timeunit;
import java.util.concurrent.atomic.atomicinteger;

/**
 * 生产者线程，负责添加队列
 */
class producerthread implements runnable {

    private blockingqueue<string> blockingqueue;

    private volatile boolean flag = true;

    private atomicinteger atomicinteger = new atomicinteger();

    producerthread(blockingqueue<string> blockingqueue) {
        this.blockingqueue = blockingqueue;
    }

    @override
    public void run() {
        try {
            system.out.println("---生产者线程启动成功---");
            while (flag) {
                string data = atomicinteger.incrementandget() + "";
                boolean success = blockingqueue.offer(data, 2, timeunit.seconds);
                if (success) {
                    system.out.println("---生产者存入队列成功->data:" + data + "---");
                } else {
                    system.out.println("---生产者存入队列失败->data:" + data + "---");
                }
                thread.sleep(1000);
            }
        } catch (interruptedexception e) {
            e.printstacktrace();
        } finally {
            system.out.println("---生产者线程已经结束---");
        }
    }

    public void stop() {
        this.flag = false;
    }

}

/**
 * 消费者线程，负责获取队列
 */
class consumerthread implements runnable {

    private blockingqueue<string> blockingqueue;

    private boolean flag = true;

    consumerthread(blockingqueue<string> blockingqueue) {
        this.blockingqueue = blockingqueue;
    }

    @override
    public void run() {
        try {
            system.out.println("---消费者线程启动成功---");
            while (flag) {
                string data = blockingqueue.poll(2, timeunit.seconds);
                if (data == null) {
                    system.out.println("---消费者没有获取到队列信息---");
                    flag = false;
                    return;
                }
                system.out.println("---消费者获得队列信息->data:" + data + "---");
            }
        } catch (interruptedexception e) {
            e.printstacktrace();
        } finally {
            system.out.println("---消费者线程已经结束---");
        }
    }

}

public class test {
    public static void main(string[] args) {
        try {
            blockingqueue<string> blockingqueue = new linkedblockingqueue<>(10);
            producerthread producerthread = new producerthread(blockingqueue);
            consumerthread consumerthread = new consumerthread(blockingqueue);
            thread producer = new thread(producerthread);
            thread consumer = new thread(consumerthread);
            producer.start();
            consumer.start();
            thread.sleep(10000);
            producerthread.stop();
        } catch (interruptedexception e) {
            e.printstacktrace();
        }
    }
}

打印如下：

---消费者线程启动成功---
---生产者线程启动成功---
---生产者存入队列成功->data:1---
---消费者获得队列信息->data:1---
---生产者存入队列成功->data:2---
---消费者获得队列信息->data:2---
.............................................
---生产者存入队列成功->data:9---
---消费者获得队列信息->data:9---
---生产者存入队列成功->data:10---
---消费者获得队列信息->data:10---
---生产者线程已经结束---
---消费者没有获取到队列信息---
---消费者线程已经结束---