夜雨聆风
OpenClaw.NET 技术解读(第七篇):消息管道与中间件
摘要：入站聊天、Webhook、定时任务与 HA/MQTT 事件桥（第六篇）最终都汇入 MessagePipeline 的有界 Channel；GatewayInboundMessageWorker 多路消费后，先过 MiddlewarePipeline（限流、Token/成本预算），再进入会话与 AgentRuntime。本篇对应计划中的 07，衔接第三篇 Gateway Worker 与第四篇 Core。
一、MessagePipeline：入站与出站
MessagePipeline 使用 System.Threading.Channels 维护 InboundMessage 与 OutboundMessage 两个有界队列，默认容量 1024，满时 Wait 形成背压，避免内存无界增长。
public MessagePipeline(int capacity = 1024, ILogger? logger = null){var opts = new BoundedChannelOptions(capacity){FullMode = BoundedChannelFullMode.Wait,SingleReader = false,SingleWriter = false};_inbound = Channel.CreateBounded<InboundMessage>(opts);_outbound = Channel.CreateBounded<OutboundMessage>(opts);_logger = logger;}public ChannelWriter<InboundMessage> InboundWriter => _inbound.Writer;public ChannelReader<InboundMessage> InboundReader => _inbound.Reader;public ChannelWriter<OutboundMessage> OutboundWriter => _outbound.Writer;public ChannelReader<OutboundMessage> OutboundReader => _outbound.Reader;
关闭时 DisposeAsync 会完成 Writer 并 排空 Reader 中未处理消息并记警告日志。
二、GatewayWorkers：谁在读入站
GatewayWorkers.Start启动 GatewayInboundMessageWorker 与 GatewayOutboundDeliveryWorker 等。入站 Worker 使用 pipeline.InboundReader，与 middlewarePipeline、agentRuntime、channelAdapters 等一并传入。
new GatewayInboundMessageWorker().Start(lifetime,logger,workerCount,isNonLoopbackBind,sessionManager,sessionLocks,lockLastUsed,pipeline,middlewarePipeline,wsChannel,agentRuntime,channelAdapters,config,cronScheduler,heartbeatService,toolApprovalService,approvalAuditStore,pairingManager,commandProcessor,operations,runtimeMetrics,learningService,automationService,contractGovernance,audioTranscriptionService);
GatewayInboundMessageWorker 内层循环为 WaitToReadAsync / TryRead，多 worker 并行消费同一 Reader（Channel 配置 SingleReader = false）。
三、MiddlewarePipeline 与 MessageContext
MiddlewarePipeline 按顺序执行 IMessageMiddleware；任一中间件可调用 MessageContext.ShortCircuit，则 ExecuteAsync 返回 false，后续不再进入 Agent。
public sealed class MiddlewarePipeline{private readonly IReadOnlyList<IMessageMiddleware> _middleware;publicMiddlewarePipeline(IReadOnlyList<IMessageMiddleware> middleware){_middleware = middleware;}publicasync ValueTask<bool> ExecuteAsync(MessageContext context, CancellationToken ct){if (_middleware.Count == 0)return true;var index = 0;ValueTask Next(){if (context.IsShortCircuited)return ValueTask.CompletedTask;if (index < _middleware.Count){var mw = _middleware[index++];return mw.InvokeAsync(context, Next, ct);}return ValueTask.CompletedTask;}await Next();return !context.IsShortCircuited;}}
MessageContext 携带 ChannelId / SenderId / Text、会话级 Token 累计（供预算中间件读取）及 ShortCircuitResponse。
网关在 CreateMiddlewarePipeline 中按需加入 RateLimitMiddleware（SessionRateLimitPerMinute > 0）与 TokenBudgetMiddleware（SessionTokenBudget 与可选成本回调）。
四、典型中间件行为
限流：RateLimitMiddleware 按 (ChannelId, SenderId) 维护一分钟滑动窗口，超限则 ShortCircuit 提示用户放慢发送。
Token / 成本预算：TokenBudgetMiddleware 比较 SessionInputTokens + SessionOutputTokens 与配置上限；若注入 costChecker，还可按契约治理拦截超额 USD 成本。
五、与 Agent、事件桥的关系
- 第六篇
中的 HA/MQTT 桥向 InboundWriter写入InboundMessage，与 WebSocket/渠道入站 同源。 - 第五篇
AgentRuntime在 Worker 内于会话锁内调用；管道层不解析工具，只负责 削峰、策略短路、出站回写。
延伸阅读
GatewayOutboundDeliveryWorker
（出站投递）
版权声明：基于开源仓库 OpenClaw.NET 整理，仅供技术交流。