企业官网建设流程全解析

进程内、线程间高性能无锁消息队列框架，解决传统阻塞锁队列延迟高、吞吐差、GC 卡顿的痛点。有界环形缓冲区 + 无锁生产者消费者模型，只做同进程线程数据交换。单机单链路可达数千万消息 / 秒，延迟低至几十纳秒，同等流水线吞吐是传统锁队列 8 倍，延迟低 3 个数量级。

• 数组环形存储：连续内存，CPU 缓存命中率极高；
• 预分配全部事件槽位：运行时不新建对象，几乎零 GC；
• 无锁 CAS 代替互斥锁：只用原子操作 + 内存屏障，无内核切换；
• 缓存行填充：隔离读写序号，解决伪共享；
• 单写原则：同一内存位置仅一个线程写入，减少缓存失效；
• 游标（Sequence）代替头尾指针：只移动数字，不挪动缓冲区数据。

分两种生产模式：

1. SingleProducerSequencer：单生产者，性能最优，无多线程竞争；
2. MultiProducerSequencer：多生产者，CAS 争抢写入序号，支持多线程并发打日志 / 采集。 职责：分配可用写入序号、判断缓冲区满 / 空、协调生产者与消费者游标。

消费者两种模型（关键区分）

1. IEventHandler<T> 广播消费（复制数据流）

• 所有 Handler 接收全部事件，互不干扰；
• HandleEventsWith(A,B)：A、B 并行消费同一条消息；
• .Then(C)：A、B 全部处理完，才执行 C，串行流水线； 适用：日志同时写文件 + 上报监控、数据同时存储 + 实时计算。

2. IWorkHandler<T> 负载均衡消费（分流）

• 一组 WorkHandler均分事件，一条消息只被其中一个线程处理；
• IO 密集场景（文件刷盘、数据库写入）分摊压力；
• 不广播，无多副本，节省内存。

Then () 流水线机制（单独梳理）

HandleEventsWith().Then()构建阶段串行、阶段内并行的 DAG 依赖图：

1. disruptor.HandleEventsWith(P1,P2)：阶段 1，P1、P2 并行处理所有事件；
2. .Then(F1)：阶段 2，必须等待 P1、P2 全部处理完毕，F1 才执行；
3. .Then(W1,W2)：阶段 3，W1、W2 负载均衡分流消费； 时序严格有序，同一事件按阶段依次流转，屏障保证时序不混乱。

四大等待策略（WaitStrategy，平衡延迟与 CPU）

消费者无新事件时的阻塞 / 自旋策略，按需选型：

1. SpinWaitStrategy：纯自旋，无休眠，纳秒级低延迟；CPU 占用高；适合运动控制、高频采集实时系统；
2. YieldingWaitStrategy：自旋 + 线程让出 CPU；兼顾延迟与 CPU，日志 / 工控通用首选；
3. BlockingWaitStrategy：锁阻塞，空闲线程休眠；CPU 极低，延迟高；后台日志、低吞吐服务；
4. TimeoutBlockingWaitStrategy：带超时阻塞，防止永久等待。IWorkHandler

.Then()可传入IWorkHandler形成工作池；HandleEventsWith()只接收IEventHandler（广播模式）。

disruptor // 阶段1：广播预处理（所有事件都走ParseHandler） .HandleEventsWith(new ParseHandler()) // 阶段2：工作池负载均衡分流，2个线程分摊写入任务 .Then( new LogWriteWorkHandler("info1.log"), new LogWriteWorkHandler("info2.log") );

FilterHandler 实现（IEventHandler）

Disruptor 流水线所有阶段共用同一块事件内存，上游修改ref LogEvent会传递到下游； 过滤时把Message置空，下游写入处理器判断Message.IsEmpty直接跳过 IO 操作，实现 “逻辑丢弃”。

完整流水线组装（HandleEventsWith + Then 串联过滤）

// 构造Disruptor var disruptor = new Disruptor<LogEvent>( 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 TaskCompletionSource<bool> Tcs { get; set; }// = new TaskCompletionSource<bool>(); } public class SetLogFinishedHandler : IEventHandler<LogInfo> { 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 Disruptor<LogInfo>( 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 TaskCompletionSource<bool>(); ringBuffer.Publish(seq); var res = await logInfo.Tcs.Task; // 等待日志处理完成 }