Disruptor入门

进程内、线程间高性能无锁消息队列框架解决传统阻塞锁队列延迟高、吞吐差、GC 卡顿的痛点。有界环形缓冲区 无锁生产者消费者模型只做同进程线程数据交换。单机单链路可达数千万消息 / 秒延迟低至几十纳秒同等流水线吞吐是传统锁队列 8 倍延迟低 3 个数量级。数组环形存储连续内存CPU 缓存命中率极高预分配全部事件槽位运行时不新建对象几乎零 GC无锁 CAS 代替互斥锁只用原子操作 内存屏障无内核切换缓存行填充隔离读写序号解决伪共享单写原则同一内存位置仅一个线程写入减少缓存失效游标Sequence代替头尾指针只移动数字不挪动缓冲区数据。分两种生产模式SingleProducerSequencer单生产者性能最优无多线程竞争MultiProducerSequencer多生产者CAS 争抢写入序号支持多线程并发打日志 / 采集。 职责分配可用写入序号、判断缓冲区满 / 空、协调生产者与消费者游标。消费者两种模型关键区分1. IEventHandlerT 广播消费复制数据流所有 Handler 接收全部事件互不干扰HandleEventsWith(A,B)A、B 并行消费同一条消息.Then(C)A、B 全部处理完才执行 C串行流水线 适用日志同时写文件 上报监控、数据同时存储 实时计算。2. IWorkHandlerT 负载均衡消费分流一组 WorkHandler均分事件一条消息只被其中一个线程处理IO 密集场景文件刷盘、数据库写入分摊压力不广播无多副本节省内存。Then () 流水线机制单独梳理HandleEventsWith().Then()构建阶段串行、阶段内并行的 DAG 依赖图disruptor.HandleEventsWith(P1,P2)阶段 1P1、P2 并行处理所有事件.Then(F1)阶段 2必须等待 P1、P2 全部处理完毕F1 才执行.Then(W1,W2)阶段 3W1、W2 负载均衡分流消费 时序严格有序同一事件按阶段依次流转屏障保证时序不混乱。四大等待策略WaitStrategy平衡延迟与 CPU消费者无新事件时的阻塞 / 自旋策略按需选型SpinWaitStrategy纯自旋无休眠纳秒级低延迟CPU 占用高适合运动控制、高频采集实时系统YieldingWaitStrategy自旋 线程让出 CPU兼顾延迟与 CPU日志 / 工控通用首选BlockingWaitStrategy锁阻塞空闲线程休眠CPU 极低延迟高后台日志、低吞吐服务TimeoutBlockingWaitStrategy带超时阻塞防止永久等待。IWorkHandler.Then()可传入IWorkHandler形成工作池HandleEventsWith()只接收IEventHandler广播模式。disruptor // 阶段1广播预处理所有事件都走ParseHandler .HandleEventsWith(new ParseHandler()) // 阶段2工作池负载均衡分流2个线程分摊写入任务 .Then( new LogWriteWorkHandler(info1.log), new LogWriteWorkHandler(info2.log) );FilterHandler 实现IEventHandlerDisruptor 流水线所有阶段共用同一块事件内存上游修改ref LogEvent会传递到下游 过滤时把Message置空下游写入处理器判断Message.IsEmpty直接跳过 IO 操作实现 “逻辑丢弃”。完整流水线组装HandleEventsWith Then 串联过滤// 构造Disruptor var disruptor new DisruptorLogEvent( eventFactory: () new LogEvent(), ringBufferSize: 4096, taskScheduler: TaskScheduler.Default, producerType: ProducerType.Multi, waitStrategy: new BlockingWaitStrategy() ); // 流水线链路填充时间戳 → 过滤日志 → 多线程写入文件 disruptor // 阶段1预处理补全时间戳 .HandleEventsWith(new LogFormatHandler()) // 阶段2过滤处理器串行执行所有事件必须经过过滤 .Then(new LogFilterHandler(LogLevel.Info)) // 阶段3负载均衡写入自动跳过被过滤日志 .Then( new LogWriteWorkHandler(app_1.log), new LogWriteWorkHandler(app_2.log) ); disruptor.Start();实现生产者异步等待消费者完成在有多个消费者的情况下需要在指定消费者的过程中在最后一个串行消费者中设置LogInfo类的TaskCompletionSource Tcs成员的状态为完成生产者通过Tcs来异步等待注意Tcs对象无法复用需要每次入队创建一个新的实例如每秒超过1000次请求需要做额外处理public class LogInfo { //其他成员 public TaskCompletionSourcebool Tcs { get; set; }// new TaskCompletionSourcebool(); } public class SetLogFinishedHandler : IEventHandlerLogInfo { public void OnEvent(LogInfo evt, long seq, bool endBatch) { evt.Tcs.SetResult(true); Console.WriteLine($SetLogFinishedHandler -- {evt.Message} ThreadID{Thread.CurrentThread.ManagedThreadId}); } } private void CreateDisruptor() { _disruptor new DisruptorLogInfo( eventFactory: () new LogInfo(), ringBufferSize: 1024, taskScheduler: TaskScheduler.Default, producerType: ProducerType.Single, // 单生产者 waitStrategy: new BlockingWaitStrategy() // 低CPU阻塞等待 ); _disruptor.HandleEventsWith(new LogHandler()) .Then(...) .Then(...) .Then(new SetLogFinishedHandler()); _disruptor.Start(); } private async Task SendLogEvent() { var ringBuffer _disruptor.RingBuffer; var seq ringBuffer.Next(); var logInfo ringBuffer[seq]; logInfo.Id _id; logInfo.Message $Log message {_id}; logInfo.Level (int)(_id % 5); logInfo.CreateTime DateTime.Now; //每次创建一个新的Tcs对象如每秒执行上千次造成gc压力需要优化 logInfo.Tcs new TaskCompletionSourcebool(); ringBuffer.Publish(seq); var res await logInfo.Tcs.Task; // 等待日志处理完成 }