1.为什么要用高性能队列Disruptor
为什么要说Disruptor？java SDK提供了2个有界队列
ArrayBlockQueue，LinkedBlockingQueue，基于ReentrantLock锁实现，在高并发情况下，锁的效率不高，更好的替代品有木有？Dosritpr
2.Disruptor介绍
性能更高的有界队列
Log4j2，Spring Messageing，HBase，Storm都用了Disruptor
为什么性能高？
1.内存分配连续，使用RingBuffer数据结构，数组元素初始化时一次性创建，缓存命中率更高；对象循环引用，避免频繁GC
2.避免伪共享，提升缓存利用率。
3.无锁算法，避免频繁加锁，解锁的性能消耗。
4.支持批量消费，消费者可以无锁消费多个消息
3.Disruptor 使用
- 生产者生产的对象称为Event，使用Disruptor必须定义Event，实例代码中为LongEvent
- 构建Disruptor要指定度列大小（bufferSize），还需要传入EventFactory，示例代码为LongEvent::new
- 消费Event需要通过handleEventsWith()方法注册一个事件处理器，发布Event则需要通过publishEvent()放阿飞
伪代码
//自定义Event
class LongEvent {
private long value;
public void set(long value) {
this.value = value;
}
}
//指定RingBuffer大小,
//必须是2的N次方
int bufferSize = 1024;
//构建Disruptor
Disruptor<LongEvent> disruptor
= new Disruptor<>(
LongEvent::new,
bufferSize,
DaemonThreadFactory.INSTANCE);
//注册事件处理器
disruptor.handleEventsWith(
(event, sequence, endOfBatch) ->
System.out.println("E: "+event));
//启动Disruptor
disruptor.start();
//获取RingBuffer
RingBuffer<LongEvent> ringBuffer
= disruptor.getRingBuffer();
//生产Event
ByteBuffer bb = ByteBuffer.allocate(8);
for (long l = 0; true; l++){
bb.putLong(0, l);
//生产者生产消息
ringBuffer.publishEvent(
(event, sequence, buffer) ->
event.set(buffer.getLong(0)), bb);
Thread.sleep(1000);
}
测试代码
@RunWith(SpringRunner.class)
@SpringBootTest(classes = FamaMonitorApplication.class)
public class DisruptorTest {
class MessageEvent {
public String message;
}
private static final int BUFFER_SIZE = 1024;
@Test
public void test(){
// 构建disruptor
Disruptor<MessageEvent> disruptor = new Disruptor<MessageEvent>(
MessageEvent::new,
BUFFER_SIZE,
DaemonThreadFactory.INSTANCE
);
// 注册消费者，并编写消费逻辑
disruptor.handleEventsWith(((event, sequence, endOfBatch) -> {
Thread.sleep(1000);
System.out.println("Message : " + event.message);
}));
// 启动
disruptor.start();
// 获取队列,然后生产
RingBuffer<MessageEvent> ringBuffer = disruptor.getRingBuffer();
for (int i = 0; true; i++) {
int finalI = i;
ringBuffer.publishEvent(((event, sequence, args) -> {
event.message = finalI + "";
}));
Thread.yield();
}
}
}
4.Disruptor 提升性能的秘籍1-内存连续分配
ArrayBlockQueue底层使用数组+ReentrantLock锁实现。
RingBuffer本质也是数组，这一点性能没有什么区别，区别在于Disruptor对于RingBuffer基础上做了很多优化，其中重要一项就是内存连续分配
5.程序的局部性原理
一段时间内程序的执行会限定在一个局部范围内
时间局限性：程序的某条指令一旦被执行，不久之后，这条指令可能再次被执行，某条数据被访问，不就之后这条数据可能再次被访问。
空间局限期：某块内存一旦被访问，不久之后这块内存附近的内存也有可能被访问。
6.cpu寄存器，cpu缓存
cpu缓存说的就是cpu的寄存器，cpu从内存中加载数据x时，会将x缓存在寄存器里，实际上cpu缓存x的同时，还缓存了x周围的数据，（根据程序的空间局限期，周围数据也有可能被访问）。程序能够体现局部性原理，就可以更好的利用cpu寄存器，从而提升性能。
cpu寄存器的读取速度和内存读取速度不是一个数量级。
7.ArrayBlockQueue有没有利用局部性原理，提升cpu缓存命中率？
并没有
每次队列中添加元素，创建对象E，这个对象都是由生产者线程创建的，由于创建这些元素的时间是离散不连续的，这些元素的内存地址大概率不连续
8.Disruptor利用空间局限期原理，提升了cpu缓存命中率
RingBuffer内部也是数组，但数组中的元素都是在初始化时一次性创建的，这些元素的内存地址大概率是连续的
for (int i=0; i<bufferSize; i++){
//entries[]就是RingBuffer内部的数组
//eventFactory就是前面示例代码中传入的LongEvent::new
entries[BUFFER_PAD + i]
= eventFactory.newInstance();
}
9.数组中的元素内存地址连续就能提升性能么？
对，可以
消费线程在消费队列时，遵循空间局限性原则，消费完第一个元素，很快就会消费第二个元素。消费第一个元素E1的时候，cpu会把内存E1后面的数据也加载进cpu寄存器里。如果第二个元素E2和E1的内存地址连续的。E2也会被加进Cache中。消费E2的时候，E2已经在cpu寄存器中了，不需要从内存再加载一遍，大大提升了性能了就
10.生产者生产时，不是往队列中add，而是修改队列中的元素对象
publishEvent()发布Event时，并不创建新Event，而是event.set()修改event
RingBuffer创建的event是可以循环利用的。这样还能避免频繁创建，删除Event导致频繁GC问题
11.避免伪共享，避免缓存不可用也同样重要
伪共享的存在，使得cache失效
cpu内存的缓存是按照缓存行(Cache Line)管理的，缓存行的大小通常是64字节；cpu从内存加载数据x，会同时加载x后面的(64-size(x))的数据。
Ar'rayBlockQueue的内部结构
/** 队列数组 */
final Object[] items;
/** 出队索引 */
int takeIndex;
/** 入队索引 */
int putIndex;
/** 队列中元素总数 */
int count;
cpu从内存加载takeIndex时，会连带将putIndex和count都加载进cache。
线程A运行在cpu1核上，入队操作，入队会修改putIndex，修改putIndex会导致所有核的缓存均实效，此时线程B在cpu2核上执行出队，需要获取takIndex，takeIndex缓存已实效，所以必须重新从内存里再拉取一遍，性能就下来了。。
12.taskIndex缓存为什么会失效？
入队操作不会修改takeIndex，但是takeIndex和putIndex共享一个缓存行，导致出队操作利用不了缓存，这就是伪共享带来的问题，性能并没有带来替身。即使数组中的对象内存地址是连续的。
13.如何避免伪共享带来的缓存不可用情况出现？
每个变量独占一个缓存行，不共享缓存行就可以
putIndex是一个缓存行
takeIndex是另一个缓存行
如何保证putIndex是一个缓存行呢？
在taskIndex的前后各填充56个字节，这样就能保证takeIndex独占一个缓存行
//前：填充56字节
class LhsPadding{
long p1, p2, p3, p4, p5, p6, p7;
}
class Value extends LhsPadding{
volatile long value;
}
//后：填充56字节
class RhsPadding extends Value{
long p9, p10, p11, p12, p13, p14, p15;
}
class Sequence extends RhsPadding{
//省略实现
}
14.综上所述
数组中连续的内存地址对象+排除伪共享情况，
就可以充分的利用cpu的缓存提升性能
以上代码设计上面的改动，说白了就是更极端的压榨硬件cpu缓存的性能
15.无锁算法
ArrayBlockQueue中使用了ReentrantLock锁，高并发情况下，性能不高
而Disruptor中采用无锁算法，没有加锁，解锁的性能消耗影响
看上去有点复杂，目前没看太明白，总之性能要优于使用ReentrantLock锁
基本逻辑是:如果没有足够的空余位置，就出让 CPU 使用权，然后重新计算；反之则用 CAS 设置入队索引
//生产者获取n个写入位置
do {
//cursor类似于入队索引，指的是上次生产到这里
current = cursor.get();
//目标是在生产n个
next = current + n;
//减掉一个循环
long wrapPoint = next - bufferSize;
//获取上一次的最小消费位置
long cachedGatingSequence = gatingSequenceCache.get();
//没有足够的空余位置
if (wrapPoint>cachedGatingSequence || cachedGatingSequence>current){
//重新计算所有消费者里面的最小值位置
long gatingSequence = Util.getMinimumSequence(
gatingSequences, current);
//仍然没有足够的空余位置，出让CPU使用权，重新执行下一循环
if (wrapPoint > gatingSequence){
LockSupport.parkNanos(1);
continue;
}
//从新设置上一次的最小消费位置
gatingSequenceCache.set(gatingSequence);
} else if (cursor.compareAndSet(current, next)){
//获取写入位置成功，跳出循环
break;
}
} while (true);
16.jvm对于伪代码共享的支持
@sun.misc.Contended
注解可以轻松避免伪共享(前提是jvm参数-XX:-RestrictContended)，以牺牲内存为代价。