队列是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效率低下,反而不如有锁操作
网友评论