Disruptor架构哲学
架构哲学系列旨在用最精简的话讲解高性能框架最高级的玩法，让你可以将这些设计思想为己所用。
图是坐早班车画的，字是坐晚班车写的。
介绍
Disruptor是LMAX开发的高性能无锁并发框架，其设计颠覆了传统队列模型，单线程可达每秒600万订单处理，延迟低至微秒级。
接下来，我会尽量多画图，少贴代码，少写字，这样能将架构抽象化，也能让你们明白各个组件中的关联关系。
架构设计
生产消费架构模型：生产者、环形缓冲区、序列号协调机制、消费者。
发布消费时序图
事件流设计
我在想如何介绍Disruptor的高级玩法，如果从组件层面讲，关联性不强；如果从流程层面讲，不够抽象，如果都讲，又过于冗杂，所以我默认你已经对Disruptor的使用已经熟练于心，因此我决定选择一个使用了Disruptor的成熟框架（Jraft一致性框架），从Jraft流程切入，点对点来阐述这些高性能设计。
初始化Disruptor
Jraft的Node节点在启动时会初始化Disruptor、RingBuffer，不同的是，我上面画的草图，是一单生产者、单消费者模式，这里是多生产者模式。
this.applyDisruptor = DisruptorBuilder.<LogEntryAndClosure> newInstance() //.setRingBufferSize(this.raftOptions.getDisruptorBufferSize()) // 默认16384.setEventFactory(new LogEntryAndClosureFactory()) // 事件工厂.setThreadFactory(new NamedThreadFactory("JRaft-NodeImpl-Disruptor-", true)) // 线程工厂.setProducerType(ProducerType.MULTI) // 多生产者.setWaitStrategy(new BlockingWaitStrategy()) // 阻塞等待策略.build();
this.applyDisruptor.handleEventsWith(new LogEntryAndClosureHandler()); // 事件处理器
this.applyDisruptor.setDefaultExceptionHandler(new LogExceptionHandler<Object>(getClass().getSimpleName()));
this.applyQueue = this.applyDisruptor.start();
环形队列
环形缓冲区RingBuffer，这种设计理念，很多地方都有被用到，比如常见的时间轮算法，那这种长度为2^n环形队列有什么好处呢？
- 固定内存分配：初始化时创建所有事件对象，运行时复用对象，消除GC停顿
- 循环覆盖机制：新数据覆盖旧数据，无需移动已有数据，减少内存拷贝操作
- O1时间复杂度：读写操作仅需移动指针（通过取模运算），无需遍历
- 数据预读：数据内存紧凑，依靠cpu缓存行，提高读性能
- 读写分离：可以通过两个读写指针分别操作，减少竞争
申请序列号
Jraft在数据写入的时候（如果当前节点是leader节点），会同步磁盘日志 - 多数心跳探活 - 日志提交。在这个流程之前，首先会通过Disruptor的RingBuffer#Sequencer申请一个序列号。
com.lmax.disruptor.SingleProducerSequencer#next(int)
此时就会有几个问题？
- 申请是否存在并发？
- 是否可以无限申请？
围绕这两个问题，我们来看看LMAX团队是如何解决这两个问题的。
申请是否存在并发？
单生产者模式，不会存在并发；多生产者模式，存在并发。
首先，申请序列号流程可以理解为2pc，第一阶段 - 申请，第二阶段 - 提交。
多生产者模式下，会用游标cursor来把控进度，cursor为已经提交的最后一个节点指针，next为申请的指针，所以只有当【cursor + 1 = next】时，才可以提交成功。
我是真不想贴代码，但是还是贴一下吧。（代码有精简）
public long next(int n) {long current;long next;do {current = cursor.get();next = current + n;// 判断是否覆盖if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current) {long gatingSequence = Util.getMinimumSequence(gatingSequences, current);if (wrapPoint > gatingSequence) {LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?continue;}gatingSequenceCache.set(gatingSequence);// 等待游标移动到当前位置} else if (cursor.compareAndSet(current, next)) {break;}}while (true);return next;
}
是否可以无限申请？
不能，前面说过，这个是环形队列RingBuffer，无限申请，就会将未消费的数据覆盖掉。
所以，在申请流程中，加了一层Filter：查看一下消费者组中，最小的消费序列号，是否大于生产者申请的序列号。
if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue){cursor.setVolatile(nextValue); // StoreLoad fencelong minSequence;while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue))){LockSupport.parkNanos(1L); // TODO: Use waitStrategy to spin?}this.cachedValue = minSequence;
}
针对部分高级玩法，后面详细介绍，我们先来看看这个gatingSequences是怎么来的？
最小消费序列号架构
这个最小消费序列号是怎么联动的？怎么动态感知的？这个就非常有意思了。
在上面《初始化Disruptor》中，会通过Disruptor#handleEventsWith连接消费者，在初始化时，就已经为每个消费者创建了Sequence对象，并放到Sequence[]组中，最后通过SEQUENCE_UPDATER骚操作，将Sequence[]组中的对象浅copy到gatingSequences中，这样，每个消费者维护自己的Sequence，gatingSequences也能动态感知到数组中每个元素的变化。
ok，这样，我们就申请到了序列号。
获取操作对象
com.lmax.disruptor.RingBuffer#get
发布事件
总共做两件事：更新游标 & 通知消费者开始消费，这里以BlockingWaitStrategy等待策略举例，因为该策略是cpu消耗最低的，也是生产环境常用的。
com.lmax.disruptor.RingBuffer#publish(long)
更新游标 & 通知消费者开始消费
cursor.set(sequence);
waitStrategy.signalAllWhenBlocking();
看到这里，你应该就明白了：发布事件就是更新一下游标，并且唤醒线程吗，所以，肯定有让线程阻塞，进入等待队列的操作。这个操作就属于消费领域的范畴了。
public void signalAllWhenBlocking(){Loc#lock();try{Condition#signalAll();}finally{lock.unlock();}
}
消费事件
在Disruptor启动时，会触发BatchEventProcessor#processEvents操作，里面设计了消费事件的逻辑。
这个逻辑比较简单，循环递增将要消费的序列，然后交给等待策略check，校验通过，交给自定义的消费处理器执行。
long nextSequence = sequence.get() + 1L;
while (true){final long availableSequence = sequenceBarrier.waitFor(nextSequence);while (nextSequence <= availableSequence){event = dataProvider.get(nextSequence);eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);nextSequence++;}sequence.set(availableSequence);
}
这里以BlockingWaitStrategy等待策略举例，因为该策略是cpu消耗最低的，也是生产环境常用的。
优先通过游标确定是否可消费，进而获取可用的序列号。
public long waitFor(long sequence, Sequence cursorSequence, Sequence dependentSequence, SequenceBarrier barrier){long availableSequence;if (cursorSequence.get() < sequence) {lock.lock();try {while (cursorSequence.get() < sequence) {barrier.checkAlert();processorNotifyCondition.await();}} finally {lock.unlock();}}while ((availableSequence = dependentSequence.get()) < sequence) {barrier.checkAlert();ThreadHints.onSpinWait();}return availableSequence;
}
高性能技术
环形队列
内存复用，无GC；cpu缓存行机制预读；O1时间复杂度；读写分离，降低读写并发；应用级别零拷贝。
规避伪共享
通过对游标（com.lmax.disruptor.Sequence$Value#value）字段前后填充字节，它会进行左右填充，左 7x8字节=56字节，右7x8字节=56字节。加上自己共 112 + 8 = 120（15个 long 占用空间），再加上 Object 占用 8 字节，一共 128字节。
cache line 默认 64字节占一行。
所以，无论怎么取，它都能命中，不会造成伪共享问题。
class LhsPadding {protected long p1, p2, p3, p4, p5, p6, p7;
}class Value extends com.lmax.disruptor.LhsPadding {protected volatile long value;
}class RhsPadding extends com.lmax.disruptor.Value {protected long p9, p10, p11, p12, p13, p14, p15;
}
读写屏障
游标为volatile修饰，读写操作会增加隐式屏障，读写也会有开销。
如何避免读写开销？
在申请序列号时，将游标赋值给临时变量，规避对volatile修饰的变量的操作，提升性能。
如何增加显示屏障？
为了让消费者及时感知游标的最新值，通过Usafe能力赋值，显示加storeload屏障，强制刷新cpu主缓存，清楚失效队列。
额外加点性能压测数据指标：无锁无屏障 * 1；无锁有屏障 * 3；有锁 * 10；
自旋优化
disruptor是无锁并发，所以使用了大量的cas和while操作，好处是规避了线程的上下文切换，坏处是会让cpu空转。
disruptor通过LockSupport.park(1)让线程短暂休眠，减缓空转。
通过Thread.onSpinWait()减缓cpu时钟频率，减缓cpu对缓存行重复加载的请求。这里具体实现是通过句柄直接调用方法，一方面是规避反射带来的性能开销，另一个方面是兼容低版本，因为这个方法，JDK9才存在。
常用策略：
while (!condition) {if (spinCount < 1000) {Thread.onSpinWait(); // 第一阶段：用户态自旋spinCount++;} else {LockSupport.parkNanos(1); // 第二阶段：内核阻塞}
}
数组读写
通过Usafe方法直接从内存中读写对象，避开JVM层对数据越界等风险检查和函数调用。业务层不建议使用Usafe相关函数，没啥必要。
sun.misc.Unsafe#getObject(java.lang.Object, long)
sun.misc.Unsafe#putInt(java.lang.Object, long)
工程实践
背景
以我在负责的风控业务为例，风控业务覆盖 数据分析（店铺、商品、订单｜品退、品差、nps、ccr）、规则中心（风控名单、字典、权限配置）、风控决策、线索归一、处置中心、申诉中心、审核中心等，历史业务没有实现who-how-what，出现问题无法溯源，也不利于业务方对存量业务的分析和规划。
目标：尽可能减少对业务RT的影响，实现一套高性能日志收敛框架，将数据收敛至分析型数据库，以供业务溯源和分析。
设计
整体设计通过多生产者-单消费者模式，进行日志输出，读写分离架构将日志写入和处理进行解耦。
瓶颈：架构性能瓶颈在于数据库，如果单条写入，高峰期，会占用磁盘带宽，竞争I/O通道。InnoDB缓冲池（Buffer Pool）会被日志数据占据，核心业务的热点数据页被挤出内存。虽然日志表单独占用底层数据库连接池，但是应用层数据库连接池可能会扛不住，会影响正常业务。
方案：消费日志时，通过队列收敛日志，批量存入redis中，用定时任务批处理redis队列中的日志数据，批量存储至分析型数据库中。冷峰或高峰期，将队列中的数据分批存入redis中。
弊端：如果机器因为宕机可能会导致缓冲队列中的数据丢失。
解决方案：运维实现优雅下线，在服务器重启或宕机的时候，将ringbuffer中的数据和缓冲队列中的数据刷到redis中；持久化ringbuffer，但是有啥必要呢？不如直接将数据同步存到redis队列中。
代码
发布者
public class LogEventDisruptor {private static final int RING_BUFFER_SIZE = 1024 * 1024; // 2^20private static final int BATCH_SIZE = 100;private final Disruptor<LogEvent> disruptor;private final RingBuffer<LogEvent> ringBuffer;public LogEventDisruptor() {this.disruptor = new Disruptor<>(LogEventFactory::new,RING_BUFFER_SIZE,Executors.defaultThreadFactory(),ProducerType.MULTI, // 支持多生产者new YieldingWaitStrategy() // 低延迟策略);// 设置批量处理器this.disruptor.handleEventsWithWorkerPool(new LogEventHandler(BATCH_SIZE));this.ringBuffer = disruptor.start();}public void publish(LogEvent event) {long sequence = ringBuffer.next();try {LogEvent bufferEvent = ringBuffer.get(sequence);BeanUtils.copyProperties(event, bufferEvent);} finally {ringBuffer.publish(sequence);}}
}
消费者
public class RiskLogComsumer implements EventHandler<LogEvent> {private final BlockingQueue<LogEvent> batchQueue = new LinkedBlockingQueue<>(BATCH_SIZE);private final int maxBatchSize;public LogEventHandler(int maxBatchSize) {this.maxBatchSize = maxBatchSize;startBatching();}@Overridepublic void onEvent(LogEvent event, long sequence, boolean endOfBatch) {batchQueue.offer(event);if (endOfBatch || batchQueue.size() >= maxBatchSize) {List<LogEvent> batch = new ArrayList<>(batchQueue.size());batchQueue.drainTo(batch);RedisBatchWriter.getInstance().writeLogs(batch);}}private void startBatching() {ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();scheduler.scheduleAtFixedRate(() -> {if (!batchQueue.isEmpty()) {int size = batchQueue.size();List<LogEvent> batch = new ArrayList<>(size);batchQueue.drainTo(batch);int split = size % maxBatchSize + 1;List<List<LogEvent>> partition = Lists.partition(batch,split);partition.forEach(source -> RedisBatchWriter.getInstance().writeLogs(source));}}, 100, 100, TimeUnit.MILLISECONDS);}
}
Redis批量写入器
public class RedisBatchWriter {private static final String REDIS_KEY = "log_queue";private static final int BATCH_SIZE = 500;private static final int TIMEOUT = 500; // msprivate static RedisBatchWriter instance;private final JedisPool jedisPool;private RedisBatchWriter() {this.jedisPool = new JedisPool(new JedisPoolConfig(), "localhost");}public static synchronized RedisBatchWriter getInstance() {if (instance == null) {instance = new RedisBatchWriter();}return instance;}public void writeLogs(List<LogEvent> logs) {try (Jedis jedis = jedisPool.getResource()) {String jsonLogs = logs.stream().map(log -> JSON.toJSONString(log)).collect(Collectors.joining("\n"));// 使用管道提升性能Pipeline pipeline = jedis.pipelined();pipeline.rPush(REDIS_KEY, jsonLogs);pipeline.sync();}}
}
定时任务批量入库
@Component
public class LogBatchJob {@Autowiredprivate LogRepository logRepository;@Scheduled(cron = "0 */1 * * * ?") // 每分钟执行public void processLogs() {try (Jedis jedis = RedisBatchWriter.getInstance().getJedisPool().getResource()) {// 使用SCAN避免阻塞ScanParams scanParams = new ScanParams().count(1000);String cursor = "0";do {ScanResult<String> result = jedis.scan(cursor, scanParams);cursor = result.getCursor();List<String> logs = result.getResult().stream().map(JSON::parseObject).map(log -> {LogEntity entity = new LogEntity();BeanUtils.copyProperties(log, entity);return entity;}).collect(Collectors.toList());if (!logs.isEmpty()) {logRepository.saveAll(logs);jedis.del(REDIS_KEY); // 清空队列}} while (!cursor.equals("0"));}}
}