LinkedBlockingQueue和ConcurrentLi
2019-10-08 本文已影响0人
骋_listen
队列是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
这个情况下可以看出无锁操作是远高于有锁操作的
BlockingQueue-100
ConcurrentQueue-100
负载过高的情况下，CAS效率低下，反而不如有锁操作
