1. 背景
好像业界有个说法，就是在单机多线程场景下 Disruptor 号称是无敌的存在，性能可以甩 JDK 原生的多线程框架好几条街。我也是秉着探索的脚步，想要证实下这个结论。因此为验证 阻塞队列 和 Disruptor 的性能差距，这里并不做过多的严谨测试，仅仅按照 1千万、5千万、1亿三个维度的数量级作为采样测试。
首先为了将这三个数量级封装为一个抽象接口，
- 字典类
public interface Constants {
/**
* 一亿
*/
int EVENT_NUM_OHM = 100000000;
/**
* 五千万
*/
int EVENT_NUM_FM = 50000000;
/**
* 一千万
*/
int EVENT_NUM_OM = 10000000;
}
- 数据类：用以模拟真实场景下的复杂对象
@Data
@AllArgsConstructor
@NoArgsConstructor
public class DataCase implements Serializable {
private Long id ;
private String name;
}
- 有界阻塞队列：为了有个 JDK 原生性能作为参考，这里利用队列的写入和取出元素，通过一个有界阻塞队列，实现生产者和消费者。
public class ArrayBlockingQueueDemo {
public static void main(String[] args) {
ArrayBlockingQueueDemo demo = new ArrayBlockingQueueDemo();
demo.testCostTime();
}
public void testCostTime(){
// 1、手工创建线程池
ThreadPoolExecutor pool = ThreadPoolsUtil.doCreate(2,2,"disruptor");
// 2、声明队列的大小
final ArrayBlockingQueue<DataCase> queue = new ArrayBlockingQueue<>(100000000);
final long startTime = System.currentTimeMillis();
// 3、提交生产者队列
pool.submit(new ProvideRunable(queue));
// 4、提交消费者队列
pool.submit(new CoustmRunable(queue,startTime));
// 4、关闭队列
pool.shutdown();
}
class ProvideRunable implements Runnable{
private ArrayBlockingQueue<DataCase> queue;
// 3.1、构造函数，将队列引入进来
ProvideRunable(ArrayBlockingQueue<DataCase> queue){
this.queue=queue;
}
@Override
public void run() {
long i = 0;
// 3.2、循环
while(i < Constants.EVENT_NUM_FM) {
try {
// 3.3、放入元素
queue.put(new DataCase(i, "c"+i));
} catch (InterruptedException e) {
e.printStackTrace();
}
i++;
}
}
}
class CoustmRunable implements Runnable{
private ArrayBlockingQueue<DataCase> queue;
private long startTime;
// 4.1、构造函数，引入队列和开始时间
CoustmRunable(ArrayBlockingQueue<DataCase> queue,long startTime){
this.queue=queue;
this.startTime=startTime;
}
@Override
public void run() {
long k = 0;
// 4.2、遍历循环
while (k < Constants.EVENT_NUM_FM) {
try {
// 4.3、取出元素
queue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
k++;
}
long endTime = System.currentTimeMillis();
System.out.println("ArrayBlockingQueue costTime = " + (endTime - startTime) + "ms");
}
}
}
在数量为 一亿的场景下，执行耗费时间是 21830 毫秒
2. Disruptor
2.1. 核心参数
2.1.1. RingBuffer
类似环形队列的数组，随着不停地填充这个 buffer ，这个序号会一直增长，直到绕过这个环。如果想要找到数组中当前序号指向的元素，可以通过取模操作：
sequence mod array length = array index
2.1.2. Sequence
发布事件的递增序列号 value 前后各增加 7 个长整型的变量，保证 value 变量在一个单独的 64 字节的缓存行中，不会与其它变量在同一个缓存行，避免伪共享。
通过 sun.misc.Unsafe 类实现了对 value 变量的 CAS 操作。
2.1.3. Sequencer
主要针对生产者的模式，分为单生产者模式和多生产者模式。
- 单生产者模式
在获取下一个可用序列号时，不会存在多线程竞争，用长整型变量 nextSequence ，而没有使用 Sequence 从而可以提高处理速度。
- 多生产者模式
存在多个生产者同时获取下一个可用序列号的情况，存在多线程竞争，所以使用 Sequence 类型变量 cursor ，利用 Sequence 类提供的 volatile 修饰的长整型变量 value 和 Unsafe 提供的 CAS 操作，保证在多线程环境中序列号更新的线程安全，但影响了处理速度。
2.1.4. WaitStrategy
决定一个消费者如何等待生产者将 Event 置入 Disruptor 的策略，都是针对消费者线程的。
主要策略有：
- BlockingWaitStrategy：最低效且折中的策略，但其对CPU的消耗最小，并且在各种部署环境中能提供更加一致的性能表现；
- SleepingWaitStrategy：也是一种折中方案，与
BlockingWaitStrategy类似 - BusySpinWaitStrategy：可能出现当没有可用序列号时，长期占用CPU，不释放CPU使用权，导致其它线程无法获取CPU使用权。
- YieldingWaitStrategy：通过
Thread.yield方法，实现生产者和消费者之间的同步，与等待策略相比，该策略100%使用CPU - TimeoutBlockingWaitStrategy：
2.2. 性能测试
2.2.1. 生产者单核
public class DisruptorDemo {
public static void main(String[] args) {
int ringBufferSize = 65536;
// 1、自定义 ThreadFactory
ThreadFactory threadFactory = new PippinThreadFactory("disruptor");
// 2、Disruptor 构造函数
final Disruptor<DataCase> disruptor = new Disruptor<>(
() -> new DataCase(),
// 2.1 定义环形数组的大小
ringBufferSize,
threadFactory,
// 2.2 单生产者模式
ProducerType.SINGLE,
// 2.3 CPU饱和的模式
new YieldingWaitStrategy()
);
// 3、处理器
ConsumerCase consumerCase = new ConsumerCase();
disruptor.handleEventsWith(consumerCase);
disruptor.start();
// 4、手工创建线程池
ThreadPoolExecutor pool = ThreadPoolsUtil.doCreate(2,2,"Pool");
// 5、提交线程池
pool.submit(()->{
RingBuffer<DataCase> ringBuffer = disruptor.getRingBuffer();
for (long i = 0; i < Constants.EVENT_NUM_OHM; i++) {
long seq = ringBuffer.next();
DataCase data = ringBuffer.get(seq);
data.setId(i);
data.setName("c" + i);
ringBuffer.publish(seq);
}
});
pool.shutdown();
}
}
class ConsumerCase implements EventHandler<DataCase> {
private long startTime;
private int i;
public ConsumerCase() {
this.startTime = System.currentTimeMillis();
}
@Override
public void onEvent(DataCase event, long sequence, boolean endOfBatch) throws Exception {
i++;
if (i == Constants.EVENT_NUM_OHM) {
long endTime = System.currentTimeMillis();
System.out.println("Disruptor costTime = " + (endTime - startTime) + "ms");
}
}
}
