蚂蚁金服分布式链路跟踪组件 SOFATracer 中 Disruptor 实践（含源码）

1 Disruptor 简介

Disruptor 旨在在异步事件处理体系结构中提供低延迟，高吞吐量的工作队列。它确保任何数据仅由一个线程拥有以进行写访问，因此与其他结构相比，减少了写争用。目前，包括 Apache Storm、Camel、Log4j2 在内的很多知名项目都应用了 Disruptor 以获取高性能。

SOFATracer 也是基于 Disruptor 高性能无锁循环队列来提供异步打印日志到本地磁盘能力的，SOFATracer 提供两种类似的日志打印类型即摘要日志和统计日志，摘要日志：每一次调用均会落地磁盘的日志；统计日志：每隔一定时间间隔进行统计输出的日志；无论是哪种日志的输出，对于 SOFATracer 来说都需要保证较高的性能，以降低对于业务整体流程耗时的影响。

关于 Disruptor 的 一些原理分析可以参考：

Disruptor： https://ifeve.com/disruptor/

A High Performance Inter-Thread Messaging Library 高性能的线程间消息传递库。

2 案例

先通过 Disruptor 的一个小例子来有个直观的认识；先看下它的构造函数：

• eventFactory : 在环形缓冲区中创建事件的 factory；
• ringBufferSize: 环形缓冲区的大小，必须是 2 的幂；
• threadFactory：用于为处理器创建线程；
• producerType：生成器类型以支持使用正确的 sequencer 和 publisher 创建 RingBuffer；枚举类型，SINGLE、MULTI 两个项。对应于 SingleProducerSequencer 和 MultiProducerSequencer 两种 Sequencer；
• waitStrategy : 等待策略；

如果我们想构造一个 disruptor，那么我们就需要上面的这些组件。从 eventFactory 来看，还需要一个具体的 Event 来作为消息事件的载体。【下面按照官方给的案例进行简单的修改作为示例】

消息事件 LongEvent ，能够被消费的数据载体

创建消息事件的 factory

ConsumerThreadFactory

OK ，上面的这些可以满足创建一个 disruptor 了：

现在是已经有了 disruptor 了，然后通过：start 来启动：

到这里，已经构建了一个 disruptor；但是目前怎么使用它来发布消息和消费消息呢？

发布消息

下面在 for 循环中 发布 5 条数据：

消息已经发布，下面需要设定当前 disruptor 的消费处理器。前面已经有个 LongEvent 和 EventFactory ; 在 disruptor 中是通过 EventHandler 来进行消息消费的。

编写消费者代码

将 eventHandler 设置到 disruptor 的处理链上：

运行结果（这里）

3 基本概念和原理

Disruptor

整个基于 ringBuffer 实现的生产者消费者模式的容器。主要属性：

• ringBuffer：内部持有一个 RingBuffer 对象，Disruptor 内部的事件发布都是依赖这个 RingBuffer 对象完成的；
• executor：消费事件的线程池；
• consumerRepository：提供存储库机制，用于将 EventHandler 与 EventProcessor 关联起来；
• started : 用于标志当前 Disruptor 是否已经启动；
• exceptionHandler : 异常处理器，用于处理 BatchEventProcessor 事件周期中 uncaught exceptions；

RingBuffer

环形队列【实现上是一个数组】，可以类比为 BlockingQueue 之类的队列，ringBuffer 的使用，使得内存被循环使用，减少了某些场景的内存分配回收扩容等耗时操作。

• E：在事件的交换或并行协调期间存储用于共享的数据的实现 -> 消息事件；

Sequencer

RingBuffer 中生产者的顶级父接口，其直接实现有 SingleProducerSequencer 和 MultiProducerSequencer；对应 SINGLE、MULTI 两个枚举值。

EventHandler

事件处置器，改接口用于对外扩展来实现具体的消费逻辑。如上面 Demo 中的 LongEventHandler ;

• event : RingBuffer 已经发布的事件；
• sequence : 正在处理的事件的序列号；
• endOfBatch : 用来标识否是来自 RingBuffer 的批次中的最后一个事件；

SequenceBarrier

消费者路障，规定了消费者如何向下走。事实上，该路障算是变向的锁。

waitStrategy 决定了消费者采用何种等待策略。

WaitStrategy

Strategy employed for making {@link EventProcessor}s wait on a cursor {@link Sequence}.

EventProcessor 的等待策略；具体实现在 disruptor 中有 8 种：

这些等待策略不同的核心体现是在如何实现 waitFor 这个方法上。

EventProcessor

事件处理器，实际上可以理解为消费者模型的框架，实现了线程 Runnable 的 run 方法，将循环判断等操作封在了里面。该接口有三个实现类:

1、BatchEventProcessor

• xceptionHandler：异常处理器；
• DataProvider：数据来源，对应 RingBuffer；
• EventHandler：处理 Event 的回调对象；
• SequenceBarrier：对应的序号屏障；
• TimeoutHandler：超时处理器，默认情况为空，如果要设置，只需要要将关联的 EventHandler 实现 TimeOutHandler 即可；

如果我们选择使用 EventHandler 的时候，默认使用的就是 BatchEventProcessor，它与 EventHandler 是一一对应，并且是单线程执行。

如果某个 RingBuffer 有多个 BatchEventProcessor，那么就会每个 BatchEventProcessor 对应一个线程。

2、WorkProcessor

基本和 BatchEventProcessor 类似，不同在于用于处理 Event 的回调对象是 WorkHandler。

原理图

无消费者情况下，生产者保持生产，但是 remainingCapacity 保持不变。

在写 Demo 的过程中，本来想通过不设定消费者来观察 RingBuffer 可用容量变化的。但是验证过程中，一直得不到预期的结果，(注：没有设置消费者，只有生产者)，先看结果：

从结果来看，remainingCapacity 的值应该随着 发布的数量 递减的；但是实际上它并没有发生任何变化。

来看下 ringBuffer.remainingCapacity() 这个方法：

这里面又使用 sequencer.remainingCapacity() 这个方法来计算的。上面的例子中使用的是 ProducerType.SINGLE，那来看 SingleProducerSequencer 这个里面 remainingCapacity 的实现。

来解释下这段代码的含义：

假设当前 ringBuffer 的 bufferSize 是 8 ；上次申请到的序列号是 5，其实也就是说已经生产过占用的序列号是 5；假设当前已经消费到的序列号是 3，那么剩余的容量为：8-（5-2） = 5。

因为这里我们可以确定 bufferSize 和 produced 的值了，那么 remainingCapacity 的结果就取决于 getMinimumSequence 的计算结果了。

这个方法是从 Sequence 数组中获取最小序列 。如果 sequences 为空，则返回 minimum。回到上一步，看下 sequences 这个数组是从哪里过来的，它的值在哪里设置的。

gatingSequences 是 SingleProducerSequencer 父类 AbstractSequencer 中的成员变量：

gatingSequences 是在下面这个方法里面来管理的。

这个方法的调用栈向前追溯有这几个地方调用了：

WorkerPool 来管理多个消费者；hangdlerEventsWith 这个方法也是用来设置消费者的。但是在上面的测试案例中我们是想通过不设定消费者只设定生成者来观察环形队列的占用情况，所以 gatingSequences 会一直是空的，因此在计算时会把 produced 的值作为 minimum 返回。这样每次计算就相当于：

也就验证了为何在不设定消费者的情况下，remainingCapacity 的值会一直保持不变。

4 SOFATracer 中 Disruptor 实践

SOFATracer 中，AsyncCommonDigestAppenderManager 对 Disruptor 进行了封装，用于处理外部组件的 Tracer 摘要日志。该部分借助 AsyncCommonDigestAppenderManager 的源码来分析下 SOFATracer 如何使用 Disruptor 的。

SOFATracer 中使用了两种不同的事件模型，一种是 SOFATracer 内部使用的 StringEvent , 一种是外部扩展使用的 SofaTacerSpanEvent。这里以 SofaTacerSpanEvent 这种事件模型来分析。StringEvent 消息事件模型对应的是 AsyncCommonAppenderManager 类封装的 disruptor。

SofaTracerSpanEvent ( -> LongEvent)

定义消息事件模型，SofaTacerSpanEvent 和前面 Demo 中的 LongEvent 基本结构是一样的，主要是内部持有的消息数据不同，LongEvent 中是一个 long 类型的数据，SofaTacerSpanEvent 中持有的是 SofaTracerSpan 。

Consumer ( -> LongEventHandler)

Consumer 是 AsyncCommonDigestAppenderManager 的内部类；实现了 EventHandler 接口，这个 consumer 就是作为消费者存在的。

在 AsyncCommonAppenderManager 中也有一个，这个地方个人觉得可以抽出去，这样可以使得 AsyncCommonDigestAppenderManager/AsyncCommonAppenderManager 的代码看起来更干净。

SofaTracerSpanEventFactory （-> LongEventFactory）

用于产生消息事件的 Factory。

ConsumerThreadFactory (-> LongEventThreadFactory )

用来产生消费线程的 Factory。

构建 Disruptor

Disruptor 的启动委托给了 AsyncCommonDigestAppenderManager 的 start 方法来执行。

来看下，SOFATracer 中具体是在哪里调用这个 start 的：

• CommonTracerManager : 这个里面持有了 AsyncCommonDigestAppenderManager 类的一个单例对象，并且是 static 静态代码块中调用了 start 方法；这个用来输出普通日志；
• SofaTracerDigestReporterAsyncManager：这里类里面也是持有了 AsyncCommonDigestAppenderManager 类的一个单例对像，并且提供了 getSofaTracerDigestReporterAsyncManager 方法来获取该单例，在这个方法中调用了 start 方法；该对象用来输出摘要日志；

发布事件

前面的 Demo 中是通过一个 for 循环来发布事件的，在 SOFATracer 中的事件发布无非就是当有 Tracer 日志需要输出时会触发发布，那么对应的就是日志的 append 操作，将日志 append 到环形缓冲区。

SOFATracer 事件发布的调用逻辑：

追溯调用的流程，可以知道当前 span 调用 finish 时或者 SOFATracer 中调用 reportSpan 时就相当于发布了一个消息事件。

5 小结

本文对 SOFATracer 中使用 Disruptor 来进行日志输出的代码进行了简单的分析，更多内部细节原理可以自行看下 SOFATracer 的代码。SOFATracer 作为一种比较底层的中间件组件，在实际的业务开发中基本是无法感知的。但是作为技术来学习，还是有很多点可以挖一挖。

SOFATracer： https://github.com/sofastack/sofa-tracer

如果有小伙伴对中间件感兴趣，欢迎 加入我们 团队，欢迎来撩；对 SOFAStack 技术体系有兴趣的可以关注 SOFAStack： https://www.sofastack.tech/community/

本文归档在 sofastack.tech。

本文转载自公众号金融级分布式架构（ID：Antfin_SOFA）。

原文链接：

https://mp.weixin.qq.com/s/Cl30-5Ywnvp2qSN1FhLtAw