Disruptor架构哲学_disruptor 时间轮

架构哲学系列旨在用最精简的话讲解高性能框架最高级的玩法，让你可以将这些设计思想为己所用。

图是坐早班车画的，字是坐晚班车写的。

介绍

Disruptor是LMAX开发的高性能无锁并发框架，其设计颠覆了传统队列模型，单线程可达每秒600万订单处理，延迟低至微秒级。

接下来，我会尽量多画图，少贴代码，少写字，这样能将架构抽象化，也能让你们明白各个组件中的关联关系。

架构设计

生产消费架构模型：生产者、环形缓冲区、序列号协调机制、消费者。

发布消费时序图

事件流设计

我在想如何介绍Disruptor的高级玩法，如果从组件层面讲，关联性不强；如果从流程层面讲，不够抽象，如果都讲，又过于冗杂，所以我默认你已经对Disruptor的使用已经熟练于心，因此我决定选择一个使用了Disruptor的成熟框架（Jraft一致性框架），从Jraft流程切入，点对点来阐述这些高性能设计。

初始化Disruptor

Jraft的Node节点在启动时会初始化Disruptor、RingBuffer，不同的是，我上面画的草图，是一单生产者、单消费者模式，这里是多生产者模式。

this.applyDisruptor = DisruptorBuilder. 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(getClass().getSimpleName()));this.applyQueue = this.applyDisruptor.start();

环形队列

环形缓冲区RingBuffer，这种设计理念，很多地方都有被用到，比如常见的时间轮算法，那这种长度为2^n环形队列有什么好处呢？

1. 固定内存分配​：初始化时创建所有事件对象，运行时复用对象，消除GC停顿
2. 循环覆盖机制​：新数据覆盖旧数据，无需移动已有数据，减少内存拷贝操作
3. O1时间复杂度：读写操作仅需移动指针（通过取模运算），无需遍历
4. 数据预读：数据内存紧凑，依靠cpu缓存行，提高读性能
5. 读写分离：可以通过两个读写指针分别操作，减少竞争

申请序列号

Jraft在数据写入的时候（如果当前节点是leader节点），会同步磁盘日志 - 多数心跳探活 - 日志提交。在这个流程之前，首先会通过Disruptor的RingBuffer#Sequencer申请一个序列号。

com.lmax.disruptor.SingleProducerSequencer#next(int)

此时就会有几个问题？

1. 申请是否存在并发？
2. 是否可以无限申请？

围绕这两个问题，我们来看看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 fence long 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^20 private 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 { private final BlockingQueue batchQueue = new LinkedBlockingQueue(BATCH_SIZE); private final int maxBatchSize; public LogEventHandler(int maxBatchSize) { this.maxBatchSize = maxBatchSize; startBatching(); } @Override public void onEvent(LogEvent event, long sequence, boolean endOfBatch) { batchQueue.offer(event); if (endOfBatch || batchQueue.size() >= maxBatchSize) { List 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 batch = new ArrayList(size); batchQueue.drainTo(batch); int split = size % maxBatchSize + 1; List<List> 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; // ms private 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 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(); } }}

定时任务批量入库

@Componentpublic class LogBatchJob { @Autowired private 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 result = jedis.scan(cursor, scanParams); cursor = result.getCursor(); List 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\")); } }}

铁三角麦克风