XiHan Framework 处于预览阶段

查看获取方式
曦寒开发框架文档
取本页 Markdown

队列与流 ​

这一页讲消息子系统的两个能力 messaging.queue 与 messaging.stream 到底差在哪里，以及两个生产级 Provider 的行为：RabbitMQ 的持久队列、手动确认、订阅方声明队列、广播队列与死信队列；Kafka 的幂等生产者、消费组位点、分区键有序、按时刻重放、预建主题与死信主题。最后说明实例标识的配置，以及消息与 CloudEvents 协议绑定的关系。读完你能按一致性与顺序要求选 Provider，写出带分区键的发布与按时刻重放的代码。前置知识是发布与订阅。

为什么需要 ​

内存代理只在单进程内有效，进程一退出消息就没了，多实例部署下每个实例各有一份队列、彼此看不见。真正的工作队列要把消息放在独立的代理上，由多个实例竞争消费、失败后重试、重试不成进死信。工作队列之上还有一类需求是「顺序」与「回放」：同一个订单的事件必须按发生顺序处理，修复了处理器的缺陷后要把过去一小时的事件重新跑一遍。这些是流的语义，队列不承诺。框架把两类保证拆成两个能力：RabbitMQ 提供队列，Kafka 同时提供队列与流；业务代码只面对 IMessagePublisher 与 IMessageHandler，需要流的特性在清单里多声明一个能力。

队列与流的差别 ​

messaging.queue@1 只承诺至少一次与按主题消费；messaging.stream@1 在此之上多三条保证：

保证messaging.queuemessaging.stream（Kafka）
投递至少一次，按主题消费，重试到上限进死信同样
顺序不承诺；重试排到队尾，后面的消息先于重试被处理信封扩展 xihankey 决定分区，同一主题内同键的消息按发布顺序投递；重试在原位
位点代理按队列管理，消费即出队按消费组（GroupId，缺省 xihan）保存，投递完成后才提交；重启续接
重放没有IMessageReplayer.ReplayAsync(topic, since, ct) 把该时刻起的消息重新投递给当前的竞争订阅
Provider提供的能力特性入口清单 ID配置节可选绑定
Messaging.InMemorymessaging.queue@1InMemoryMessagingFeature（随 Standard 装配）xihan.messaging.inmemoryXiHan:Messaging:InMemory—
Messaging.RabbitMQmessaging.queue@1RabbitMQMessagingFeaturexihan.messaging.rabbitmqXiHan:Messaging:RabbitMQqueue
Messaging.Kafkamessaging.queue@1、messaging.stream@1KafkaMessagingFeaturexihan.messaging.kafkaXiHan:Messaging:Kafkastream

两个 Provider 的底层客户端都未声明 Native AOT 兼容，AOT 发布的应用只能用内存代理或自己的 Provider。

最小示例 ​

装配 Kafka，再写一个把同一订单的事件放进同一分区、并能按时刻重放的服务。所在特性的清单声明 messaging.stream（MessagingCapabilities.Stream），因为 IMessageReplayer 只在 Kafka 被选中提供流能力时登记。

csharp
using XiHan.Framework.Hosting.AspNetCore;
using XiHan.Framework.Messaging.Kafka;
using XiHan.Framework.Profiles.Standard;

var builder = WebApplication.CreateBuilder(args);

// Kafka 顶替内存代理提供 messaging.queue，并提供 messaging.stream
builder.AddXiHan(xihan => xihan
    .UseStandardProfile()
    .UseFeature<KafkaMessagingFeature>());

var app = builder.Build();
app.UseXiHan();
app.Run();
json
{
  "XiHan": {
    "Messaging": {
      "Kafka": {
        "BootstrapServers": "localhost:9092",
        "GroupId": "orders"
      }
    }
  }
}
csharp
using XiHan.Framework.Messaging.Abstractions.Envelopes;
using XiHan.Framework.Messaging.Abstractions.Publishing;
using XiHan.Framework.Primitives;

public sealed class OrderStream(IMessagePublisher publisher, IMessageReplayer replayer)
{
    // 同一订单的事件用订单 ID 作分区键，消费端按发布顺序收到
    public ValueTask<Result> PublishAsync(Guid orderId, MessageEnvelope envelope, CancellationToken cancellationToken)
        => publisher.PublishAsync("orders", envelope.WithExtension(EnvelopeExtensions.PartitionKey, orderId.ToString("N")), cancellationToken);

    // 修复处理器的缺陷后，把最近一小时的消息重新跑一遍；处理器要能去重
    public ValueTask<Result> ReplayLastHourAsync(CancellationToken cancellationToken)
        => replayer.ReplayAsync("orders", DateTimeOffset.UtcNow.AddHours(-1), cancellationToken);
}

本地可以用单节点容器起一个 Kafka：docker run -d -p 9092:9092 apache/kafka。启动应用后，对同一个订单连续发布三条事件，处理器按发布顺序收到它们，别的订单的事件可能交错在中间；调用 ReplayLastHourAsync 后处理器再次收到这一小时内的全部消息，正常消费的位点不受影响。主题不存在或该时刻之后没有消息时重放成功返回；代理不可达返回 MESSAGING_UNAVAILABLE_001。

RabbitMQ ​

RabbitMQMessageBroker 实现 IMessageBroker。装配后它顶替内存代理，处理器、发布者与发件箱的代码不变。

csharp
using XiHan.Framework.Hosting.AspNetCore;
using XiHan.Framework.Messaging.RabbitMQ;
using XiHan.Framework.Profiles.Standard;

var builder = WebApplication.CreateBuilder(args);

builder.AddXiHan(xihan => xihan
    .UseStandardProfile()
    .UseFeature<RabbitMQMessagingFeature>());
json
{
  "XiHan": {
    "Messaging": {
      "MaxAttempts": 5,
      "RabbitMQ": {
        "Configuration": "amqp://guest:guest@localhost:5672",
        "QueuePrefix": "orders.",
        "PrefetchCount": 32
      }
    }
  }
}
  • 发布方只声明交换机：消息发到持久的 direct 交换机 Exchange（缺省 xihan），路由键 <QueuePrefix><topic>（前缀缺省 xihan.）。发布方不声明队列。
  • 队列只由订阅方声明：订阅方启动时声明持久队列 <QueuePrefix><topic>，以同名路由键绑定到交换机。订阅方第一次启动之后，它离线期间的消息留在队列里；第一次启动之前发布的消息没有队列可进，按「没有订阅者」丢弃。希望订阅方上线前就留存消息的部署，由运维预先声明持久队列并绑定。
  • 消息持久化投递，消息体是 CloudEvents 结构化 JSON，message_id 为事件 ID。发布共用一条开启了发布确认的通道，代理确认后发布才返回成功，代理拒绝返回 MESSAGING_UNAVAILABLE_001；批量发布连续写出整批再一次等待全部确认，结果逐项对应。
  • 每个订阅（主题加方式）一个通道与一个消费者，预取 PrefetchCount（缺省 16），手动确认，单线程派发。
  • 处理失败：尝试次数（消息头 x-xihan-attempt）小于 MaxAttempts 时重新发布到所消费队列的队尾，否则发布到死信队列 <QueuePrefix><topic><DeadLetterSuffix>；重新发布同样等确认，确认了才确认原消息。无法解码的消息不重试，原样进死信队列。
  • 广播订阅：每个实例声明独占、非持久、随连接删除的队列 <QueuePrefix><topic>@<InstanceId>，绑定到同一路由键；实例下线不留下队列，重试到上限或无法解码时确认掉并记 Warning，不进死信。
  • 停止时取消消费者；正在处理的消息不确认，连接关闭后由 RabbitMQ 重新投递。
键类型缺省说明
XiHan:Messaging:RabbitMQ:Configuration字符串—AMQP URI（amqp:// 或 amqps://）；有绑定 queue 时被绑定的主值覆盖，两者都没有时装配失败；绑定的属性 username / password 覆盖 URI 里的用户名与密码
XiHan:Messaging:RabbitMQ:QueuePrefix字符串xihan.队列名与路由键的前缀
XiHan:Messaging:RabbitMQ:Exchange字符串xihan发布用的 direct 交换机，不能为空、不能以 amq. 开头；发布方与订阅方要一致
XiHan:Messaging:RabbitMQ:PrefetchCount1–6553516每个消费者的预取数
XiHan:Messaging:RabbitMQ:HealthProbeIntervalSeconds正数15健康项 messaging.rabbitmq.connection 的重探间隔，也是单次探测的超时

滚动升级时先升级订阅方，或让一个新版本的订阅实例先启动一次建好绑定，否则新主题的消息在订阅方上线前按「没有订阅者」丢弃。

Kafka ​

KafkaMessageBroker 同时实现 IMessageBroker 与 IMessageReplayer，特性只为被选中的能力绑定接口。

  • 主题映射为 <TopicPrefix><topic>（前缀缺省 xihan.）。消息值是 CloudEvents 结构化 JSON，头 content-type 为 application/cloudevents+json、x-xihan-attempt 为尝试次数。
  • 消息键取信封扩展 xihankey（EnvelopeExtensions.PartitionKey）：同键进同一分区、按发布顺序投递；没有键时由 Kafka 分配分区，只在主题只有一个分区时全局有序。
  • 生产者幂等且 acks=all。批量发布交给客户端合并成批次发出，再一次等待全部投递报告，任一失败整批抛出 MESSAGING_UNAVAILABLE_001。
  • 每个订阅（主题加方式）一个消费者，单线程按分区顺序派发。竞争订阅属于同一消费组 GroupId，没有已提交位点时从最早位点开始（订阅前发布的消息也会收到）。
  • 处理失败在原位重试到 MaxAttempts，不重新入队，分区内顺序不变；竞争订阅仍失败就发布到死信主题 <TopicPrefix><topic><DeadLetterSuffix>。无法解码的消息直接进死信。一条消息重试期间，同分区后面的消息等待。
  • 竞争订阅投递完成（成功或进死信）后才提交位点；停止时未完成的消息在重启后重新投递。死信发布没有得到确认时不提交，消费者回退到这条消息，等 RedeliveryDelaySeconds（缺省 5）后重新投递，期间这个分区不往后读。
  • 广播订阅用本实例独占的消费组 <GroupId>@<InstanceId>，从最新位点开始、不提交位点，重启不回放历史；启动后要等消费组分到分区（通常几秒）才开始收；重试耗尽或无法解码时丢弃并记 Warning，不发布死信。
  • 重放：ReplayAsync(topic, since) 只投给竞争订阅，用一个新的独立消费组，从各分区该时刻的位点读到调用时的高水位，经同样的派发、重试与死信规则再投递一遍；不提交位点、不影响正常消费；重放期间新到的消息由正常消费投递。
键类型缺省说明
XiHan:Messaging:Kafka:BootstrapServers字符串—引导地址，逗号分隔的 host:port；有绑定 stream 时被绑定的主值覆盖，两者都没有时装配失败。绑定的属性 security.protocol、sasl.mechanism、sasl.username、sasl.password 各自覆盖 Properties 里的同名键
XiHan:Messaging:Kafka:GroupId字符串xihan消费组，同一组的实例分摊分区
XiHan:Messaging:Kafka:TopicPrefix字符串xihan.主题前缀
XiHan:Messaging:Kafka:RedeliveryDelaySeconds正数5投递没有完成（死信发布失败等）时重新投递前的等待
XiHan:Messaging:Kafka:HealthProbeIntervalSeconds正数15健康项 messaging.kafka.connection 的重探间隔，也是单次探测的超时
XiHan:Messaging:Kafka:Properties:<key>字符串—原样透传给 librdkafka（生产者与消费者共用），如 security.protocol、sasl.mechanism、ssl.ca.location

凭据从密钥来源注入，不要写在仓库里的配置文件中，见远程配置与密钥。

Kafka 关了自动建主题时要预建主题

Kafka 缺省开启 auto.create.topics.enable，发布或订阅到不存在的主题时由 broker 建出来。生产集群常把它关掉，这时主题不存在是部署错误，框架不替你建，也不把它当作「没有订阅者」丢弃（那样主题名或 TopicPrefix 写错也会被吞掉）：

  • 发布约 30 秒后失败（MESSAGING_UNAVAILABLE_001）；经发件箱的按发件箱规则退避重试，到上限进死信。
  • 订阅记 Warning 日志并一直等，主题建好之后才开始收。
  • 死信主题不存在时死信发布失败，消息回退重投，这个分区停在原地。

所以要预先建好（名字都带 TopicPrefix，缺省 xihan.）：应用发布与订阅的主题、被竞争订阅主题的死信主题 <topic>.dlq，以及框架自己发布的事件主题：装配了调度时的 scheduling.job.* 与 scheduling.run.*（见作业与计划），装配了工作流时的 workflow.*（见定义与发布），没人订阅的也要建。完整清单与建主题的命令见 Messaging.Kafka。

实例标识 ​

实例标识是 MessagingInstance.Id（消息装配登记的单例，每个宿主一个），广播订阅的队列名与消费组名用它。缺省每次启动生成（机器名、进程 ID 与 8 位随机十六进制）；要在 RabbitMQ 管理界面与 Kafka 消费组列表里认出实例，可以用 XiHan:Messaging:InstanceId 配一个固定值：

json
{ "XiHan": { "Messaging": { "InstanceId": "orders-api-0" } } }
  • 文法：1 到 64 个字母、数字、.、_、-，字母或数字开头（不含 @）。写了但不合文法（含空字符串）在装配时抛 ArgumentException；不想固定就不写这个键。
  • 必须每个同时在线的实例都不同。Kafka 上两个实例同一标识就是同一个广播消费组，组内分摊分区，广播静默退化成竞争；RabbitMQ 上后启动的实例声明同名独占队列被代理拒绝，启动失败（MESSAGING_UNAVAILABLE_001）。
  • 多副本部署配成部署平台保证唯一的值，例如 Kubernetes 用下行 API 把 Pod 名设进环境变量，见 Kubernetes；多个副本共用的配置文件里不要写死。
  • 固定标识不改变广播语义：重启后仍不回放历史。代价是旧连接没有及时断开时（节点失联、进程被冻结），用同一标识重启的实例要等代理发现：RabbitMQ 在旧连接心跳超时前启动失败，Kafka 在旧成员会话超时前收不到广播。缺省的生成值每次都不同，没有这个问题。

两个 Provider 的重试之间都不等待，次数都由 XiHan:Messaging:MaxAttempts 决定；发件箱有自己独立的重试与死信规则，见领域事件与发件箱。

与 CloudEvents 协议的关系 ​

消息子系统管的是进程与代理之间的传输；Protocols.CloudEvents 是同一份信封在 HTTP 上的协议绑定：入站端点 MapXiHanCloudEvents() 接收 binary / structured / batch 三种内容模式，按 XiHan:Protocols:CloudEvents:Topics:<事件类型> 映射到主题，缺省在请求内经同一个 MessageDispatcher 派发给处理器（也可以只发布到代理）；出站的 CloudEventsWebhookPublisher 把信封推送到配置的 Webhook。framework/samples/SimpleApp 的 POST /events 就是把外部的 orders.placed 事件交给 OrderPlacedHandler。所以同一个 IMessageHandler 既能处理代理投递的消息，也能处理 HTTP 推来的 CloudEvent，细节见 CloudEvents。

健康项 ​

健康项来源Degraded 的条件
messaging.rabbitmq.connectionMessaging.RabbitMQ后台按 HealthProbeIntervalSeconds 检查连接与发布通道，断开时 Degraded（MESSAGING_UNAVAILABLE_001）
messaging.kafka.connectionMessaging.Kafka后台按 HealthProbeIntervalSeconds 描述集群，不可达或超时时 Degraded（KERNEL_UNAVAILABLE_005）

两个 Provider 启动时都先连接代理再开始消费，连不上即启动失败（MESSAGING_UNAVAILABLE_001），不退回内存代理。

常见问题 ​

RabbitMQ 和 Kafka 怎么选 ​

只要工作队列、不在意同一主题内的顺序：RabbitMQ 更轻，队列按订阅方声明、死信就是一个队列。同一业务对象的事件必须按顺序处理、或需要在修复后重放历史：Kafka，给信封设分区键，用 IMessageReplayer。单进程、开发与测试用内存代理。

订阅方还没启动过，这期间发布的消息去哪了 ​

RabbitMQ 上没有队列可进，按「没有订阅者」丢弃；订阅方第一次启动声明队列之后，它离线期间的消息才会留在队列里。要在订阅方上线之前就留存，由运维预先声明持久队列并绑定。Kafka 上消息进主题、留到保留期结束，竞争订阅第一次启动时从最早位点开始，能收到。

重放会影响正常消费吗 ​

不会。重放用一个新的独立消费组，从该时刻读到调用时的高水位，不提交位点；正常消费组照常推进。重放只投给竞争订阅，处理器会再次收到这些消息，所以处理器要能按 Id + Source 去重。

为什么 Kafka 的一条消息失败会卡住后面的消息 ​

Kafka 的重试在原位、不重新入队，这是分区内有序的代价：同分区后面的消息要等这条消息成功或进死信。重试之间不等待，MaxAttempts 缺省 3，通常只是短暂停顿；处理器对暂时性故障应快速失败，对永久性故障应尽早返回失败让它进死信。关了自动建主题的集群上收不到消息时，先检查主题名是否带了 TopicPrefix 并按上面的清单预建主题。

相关包与下一步 ​

Released under The MIT License

Copyright ©2021-Present XiHanFun and contributors.