GTLC全球技术领导力峰会·上海站,首批讲师正式上线! 了解详情
写点什么

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

2020 年 3 月 24 日

蚂蚁金服分布式链路跟踪组件 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


2020 年 3 月 24 日 10:211432

评论 2 条评论

发布
用户头像
Cusomer里面的volatile修饰的成员变量感觉有bug,volatile貌似只对值类型变量有效果,这个地方应该换成AtomicReference是不是更好?
2020 年 03 月 24 日 19:14
回复
不好意思应该SofaTracerSpanEvent
2020 年 03 月 24 日 19:20
回复
没有更多了
发现更多内容

安卓开发实战讲解!从新手到Flutter架构师,一篇就够!快来收藏!

欢喜学安卓

android 程序员 面试 移动开发

关于列表转字符串这个过程的曲折

ベ布小禅

四月日更

计算机原理学习笔记 Day2

穿过生命散发芬芳

计算机原理 4月日更

嘉云公司研发效能平台实践

小江

研发效能 CI/CD

HBase的rowKey设计技巧

五分钟学大数据

HBase 4月日更

华仔架构实战营 - 作业 - 模块2

曲元洪

架构实战营

翻译:《实用的Python编程》08_03_Debugging

codists

Python

GitHub已爆火的Java突击手册,全面详细对标P7岗!真的很全面

比伯

Java 编程 架构 程序人生 计算机

「架构师训练营 4 期」大作业一&二

凯迪

架构师训练营 4 期

构建WebRTC音视频系统处理结构

正向成长

音视频 WebRTC

并发容器与并发控制 - JUC

学Java关注我

Java 编程 程序员 架构 计算机

猫鼠游戏,一个刷票老千看在线投票项目的防范与取舍

ucsheep

安全 在线投票 防作弊 刷票

Java 并发基础(五):面试实战之多线程顺序打印

看山

Java并发

安卓开发基础面试题,分享一点面试小经验,含BATJM大厂

欢喜学安卓

android 程序员 面试 移动开发

吃透Nginx编译安装过程

书旅

nginx Nginx PHP-FPM

赛文 | 阿里JVM-Sandbox核心源码剖析

九叔

JVM 中间件 类加载 Sandbox 类隔离

算法训练营 - 学习笔记 - 第二周

心在飞

taskwarrior ,一款提升效率的命令行的 TODO list 工具

Red

效率工具 TODO linux操作

浅析LSM-Tree存储模型

正向成长

LSM树 KV存储引擎

外包3年熬出头,3面字节、6面腾讯一次过,谈谈我的大厂面经

神奇小汤圆

Java 编程 程序员 架构 面试

「架构师训练营 4 期」大作业二

凯迪

架构师训练营 4 期

陪伴

小天同学

陪伴 育儿 个人感悟 4月日更

Dubbo 学习笔记(二) Spring Boot 整合 Dubbo

U+2647

Spring Boot dubbo 四月日更

重读《重构2》

顿晓

重构 4月日更

Python OpenCV 图片高斯模糊

梦想橡皮擦

Python OpenCV 4月日更

继续探究:一文理清JVM和GC(下)

比伯

Java 架构 程序人生 计算机 技术宅

再谈日更公众号

彭宏豪95

写作 感悟 微信公众号 4月日更

PI的一种简写。

山@支

hive的数据存储格式

大数据技术指南

hive 4月日更

Docker 环境清理的常用方法

xcbeyond

Docker 4月日更

手撕83K STAR的Axios设计思想,并进行能力增强

梁龙先森

源码分析 前端进阶 axios

DNSPod与开源应用专场

DNSPod与开源应用专场

蚂蚁金服分布式链路跟踪组件 SOFATracer 中 Disruptor 实践(含源码)-InfoQ