尧图网页设计 
使用 Watermill CQRS 与 Kafka 分区键实现严格有序事件处理邮件订阅系统实战解析【免费下载链接】watermillBuilding event-driven applications the easy way in Go.项目地址: https://gitcode.com/GitHub_Trending/wa/watermill本篇指南基于仓库中的完整可运行示例 _examples/basic/6-cqrs-ordered-eventsREADME 位于 _examples/basic/6-cqrs-ordered-events/README.md讲解如何在 Go 中基于 Watermill 的 CQRS 组件构建一个事件顺序敏感的邮件订阅系统用户可订阅、更新邮箱、退订系统同时维护活跃订阅者列表与订阅活动时间线两份读模型。读完后你将掌握Watermill CQRS 组件的基本装配方法、用**事件处理组Event Group**保证同一组事件消费有序的原理以及如何利用Kafka 分区键在不牺牲吞吐量的前提下维持每个订阅者事件的相对顺序。示例应用邮件订阅系统这是一个用 Watermill CQRS 组件实现的邮件订阅系统用户可以对邮件列表执行三种操作订阅提供邮箱地址加入订阅列表更新邮箱订阅后修改自己的邮箱地址退订从订阅列表中移除。系统始终维护两份视图Read Model当前活跃订阅者列表map[subscriberID]email实时反映谁是有效订阅者订阅活动时间线按时间顺序记录每一次SUBSCRIBED、UNSUBSCRIBED、EMAIL_UPDATED活动。系统整体遵循 CQRS 模式写操作通过Command命令发出由 Command Handler 处理后发布Event事件读模型则订阅事件流并增量更新。整个数据流由 message.Router 驱动这正是 Watermill 构建事件驱动应用的通用骨架。为什么事件顺序在这里是生死攸关的README 中明确指出了顺序的关键性In this example, keeping order of events is crucial. If events wont be ordered, andSubscriberSubscribedwould arrive afterSubscriberUnsubscribedevent, the subscriber will be still subscribed.保持事件顺序至关重要。如果事件乱序SubscriberSubscribed先于SubscriberUnsubscribed被处理那么用户明明退订了订阅者列表中却仍然保留着他。以活跃订阅者列表这个读模型为例它的状态变更完全依赖事件到达的顺序事件序列列表状态Subscribed→Unsubscribed用户被正确移除Unsubscribed→Subscribed乱序用户被错误地保留在列表中同理UpdateEmail与Unsubscribe之间也存在类似竞态若邮箱更新事件晚于退订事件到达一个已退订用户的新邮箱可能被重新写入列表。因此为每个订阅者的事件流维持严格顺序是本示例的核心工程问题而答案由两个机制共同给出事件处理组Event Group让多个事件处理器共享同一个订阅者实例保证同一读模型的事件按到达顺序逐个消费Kafka 分区键Partition Key以subscriber_id作为分区键让同一订阅者的所有消息落入 Kafka 的同一分区从而在不牺牲并行吞吐量的前提下维持每个订阅者的相对顺序。架构全景从总线到处理器示例在 _examples/basic/6-cqrs-ordered-events/main.go 中完成全部装配整体分为四层Kafka (commands topic) Kafka (events topic) │ │ ▼ ▼ CommandProcessor (3 个 Handler) EventGroupProcessor (2 个 Group) │ │ ▼ ▼ CommandBus ──► CommandHandler ──► EventBus ──► Read ModelCommandBus / EventBus负责发布命令与事件写入侧CommandProcessor订阅commandstopic按事件类型把命令分发给对应 HandlerEventGroupProcessor订阅eventstopic把同一读模型的一组事件处理器绑定到同一个 Kafka 订阅上。用事件处理组保证组内有序消费普通EventProcessor会为每一个EventHandler 单独创建一个订阅者参见 components/cqrs/event_processor.go 中SubscriberConstructor的注释而EventGroupProcessor则允许多个处理器共享同一个订阅者实例。这在 components/cqrs/event_processor_group.go 的注释中有明确说明Compared to EventProcessor, EventGroupProcessor allows to have multiple handlers that share the same subscriber instance.这带来两个直接好处单个订阅消费一个读模型对应一个 Kafka 消费组ConsumerGroup消息以严格顺序到达并被逐条分发组内路由当消息到达时Watermill 会根据消息中的事件名匹配组内的对应处理器见EventGroupProcessor.routerHandlerGroupFunccomponents/cqrs/event_processor_group.go未匹配到的事件会以 trace 日志跳过。示例中注册了两个事件组_examples/basic/6-cqrs-ordered-events/main.go// 组 1订阅者读模型共享同一个订阅 err eventProcessor.AddHandlersGroup( SubscriberReadModel, cqrs.NewGroupEventHandler(subscribersReadModel.OnSubscribed), cqrs.NewGroupEventHandler(subscribersReadModel.OnUnsubscribed), cqrs.NewGroupEventHandler(subscribersReadModel.OnEmailUpdated), ) // 组 2活动时间线读模型共享同一个订阅 err eventProcessor.AddHandlersGroup( ActivityTimelineReadModel, cqrs.NewGroupEventHandler(activityReadModel.OnSubscribed), cqrs.NewGroupEventHandler(activityReadModel.OnUnsubscribed), cqrs.NewGroupEventHandler(activityReadModel.OnEmailUpdated), )每个组在创建订阅者时都使用params.EventGroupName作为 Kafka 消费组名_examples/basic/6-cqrs-ordered-events/main.go即SubscriberReadModel、ActivityTimelineReadModel各自拥有独立的消费组互不干扰地各自按序消费。EventGroupProcessor 的核心配置项根据 components/cqrs/event_processor_group.goEventGroupProcessorConfig的关键字段如下配置项是否必填作用GenerateSubscribeTopic必填为事件组生成订阅 topic示例中统一返回eventsSubscriberConstructor必填为每个事件组创建订阅者每个组只调用一次这正是一组一订阅、有序消费的根基Marshaler必填事件/命令的编解码器OnHandle可选类似中间件可在事件处理前后注入逻辑须显式调用params.Handler.Handle()AckOnUnknownEvent可选事件无对应处理器时是否 ack默认不 ack 并报错Logger可选默认watermill.NopLogger单一 Topic 分区键顺序与吞吐兼得全局单一 topic 的意义为了让所有订阅者的事件流共享同一个有序消费序列命令与事件分别被路由到单一的全局 topic_examples/basic/6-cqrs-ordered-events/main.gocommandBus, err : cqrs.NewCommandBusWithConfig(publisher, cqrs.CommandBusConfig{ GeneratePublishTopic: func(params cqrs.CommandBusGeneratePublishTopicParams) (string, error) { // 所有命令共用一个 topic以保持命令的有序性 return commands, nil }, Marshaler: cqrsMarshaler, Logger: logger, }) eventBus, err : cqrs.NewEventBusWithConfig(publisher, cqrs.EventBusConfig{ GeneratePublishTopic: func(params cqrs.GenerateEventPublishTopicParams) (string, error) { // 所有事件共用一个 topic以保持事件的有序性 return events, nil }, Marshaler: cqrsMarshaler, Logger: logger, })这里用GeneratePublishTopic回调把 topic 固定为常量而不是按事件/命令类型生成对比cqrs.StructName按类型名命名 topic 的默认策略。为什么 Kafka 分区键是关键Kafka 保证同一分区内消息有序。如果所有消息都挤在一个分区虽然全局有序但吞吐量会被单分区上限卡死如果消息随机分散到多个分区顺序又无法保证。示例的解法是以subscriber_id作为分区键。同一订阅者的消息订阅、更新、退订永远落入同一分区Kafka 保证它们在分区内有序而不同订阅者的消息可以并行分布在不同分区由多个消费者并发处理——既保序又不损失吞吐。这正是 README 中使用 Kafka 分区键在保持事件顺序的同时提升处理吞吐的实现方式。在 _examples/basic/6-cqrs-ordered-events/main.go 中通过 watermill-kafka 提供的kafka.NewWithPartitioningMarshaler启用分区键kafkaMarshaler : kafka.NewWithPartitioningMarshaler(GenerateKafkaPartitionKey)GenerateKafkaPartitionKey从 Watermill 消息元数据中取出partition_key字段并返回_examples/basic/6-cqrs-ordered-events/message.gofunc GenerateKafkaPartitionKey(topic string, msg *message.Message) (string, error) { slog.Debug(Setting partition key, topic, topic, msg_metadata, msg.Metadata) return msg.Metadata.Get(PartitionKeyMetadataField), nil }分区键如何随消息流转分区键不会凭空产生示例通过装饰cqrs.ProtoMarshaler把 Proto 消息中的业务元数据搬运到 Watermill 消息元数据中_examples/basic/6-cqrs-ordered-events/message.gotype CqrsMarshalerDecorator struct { cqrs.ProtoMarshaler } const PartitionKeyMetadataField partition_key func (c CqrsMarshalerDecorator) Marshal(v interface{}) (*message.Message, error) { msg, err : c.ProtoMarshaler.Marshal(v) if err ! nil { return nil, err } pm, ok : v.(ProtoMessage) if !ok { return nil, fmt.Errorf(%T does not implement ProtoMessage and cant be marshaled, v) } metadata : pm.GetMetadata() if metadata nil { return nil, fmt.Errorf(%T.GetMetadata returned nil, v) } msg.Metadata.Set(PartitionKeyMetadataField, metadata.PartitionKey) msg.Metadata.Set(created_at, metadata.CreatedAt.AsTime().String()) return msg, nil }MessageMetadata是每个命令/事件的公共 Proto 字段_examples/basic/6-cqrs-ordered-events/proto/messages.proto由GenerateMessageMetadata(partitionKey string)统一创建_examples/basic/6-cqrs-ordered-events/message.go。在模拟流量中subscriber_id即分区键_examples/basic/6-cqrs-ordered-events/main.go。整条链路因此闭环simulateTraffic 生成命令 │ Metadata.PartitionKey subscriberID ▼ CqrsMarshalerDecorator 把 partition_key 写入消息元数据 ▼ kafka.NewWithPartitioningMarshaler 读取 partition_key 作为 Kafka 分区键 ▼ 同一订阅者的消息进入同一 Kafka 分区分区内有序 ▼ EventGroupProcessor 按序消费并分发到读模型处理器装配细节Router、中间件与处理器Router 与中间件CQRS 建立在 Watermill 消息路由之上components/cqrs 组件全部以*message.Router为基座。示例创建 Router 并挂载中间件_examples/basic/6-cqrs-ordered-events/main.gorouter, err : message.NewRouter(message.RouterConfig{}, logger) if err ! nil { panic(err) } // 恢复事件/命令处理器中的 panic防止消费者进程崩溃 router.AddMiddleware(middleware.Recoverer) // 自定义中间件记录每条收到的消息元数据 router.AddMiddleware(func(h message.HandlerFunc) message.HandlerFunc { return func(msg *message.Message) ([]*message.Message, error) { slog.Debug(Received message, metadata, msg.Metadata) return h(msg) } })middleware.Recoverer位于 message/router/middleware/recoverer.go是所有生产级 Watermill 应用的基础防护你还可以从 message/router/middleware 中选择重试Retry、超时Timeout、熔断CircuitBreaker、去重Deduplicator等中间件。Kafka 发布/订阅示例使用 Redpanda 提供 Kafka 兼容的 broker地址为kafka:9092。发布端_examples/basic/6-cqrs-ordered-events/main.gopublisher, err : kafka.NewPublisher( kafka.PublisherConfig{ Brokers: []string{kafka:9092}, Marshaler: kafkaMarshaler, // 带分区键的 marshaler }, watermillLogger, )订阅端由SubscriberConstructor按需创建命令处理器用params.HandlerName作为消费组名每个命令 handler 独立消费组事件组处理器用params.EventGroupName作为消费组名每个读模型一个消费组。命令处理器三个命令分别对应三个 handler_examples/basic/6-cqrs-ordered-events/main.goerr commandProcessor.AddHandlers( cqrs.NewCommandHandler(SubscribeHandler, SubscribeHandler{eventBus}.Handle), cqrs.NewCommandHandler(UnsubscribeHandler, UnsubscribeHandler{eventBus}.Handle), cqrs.NewCommandHandler(UpdateEmailHandler, UpdateEmailHandler{eventBus}.Handle), )每个 CommandHandler 的职责是把命令转换为领域事件并通过EventBus发布。例如SubscribeHandler_examples/basic/6-cqrs-ordered-events/main.gofunc (h SubscribeHandler) Handle(ctx context.Context, cmd *Subscribe) error { return h.eventBus.Publish(ctx, SubscriberSubscribed{ Metadata: GenerateMessageMetadata(cmd.SubscriberId), SubscriberId: cmd.SubscriberId, Email: cmd.Email, }) }UnsubscribeHandler与UpdateEmailHandler结构相同分别发布SubscriberUnsubscribed与SubscriberEmailUpdated。读模型纯内存 互斥锁两个读模型都定义在独立文件中订阅者列表_examples/basic/6-cqrs-ordered-events/subscribers.gomap[string]stringsubscriberID → emailOnSubscribed写入、OnUnsubscribed删除、OnEmailUpdated更新并通过sync.RWMutex保证并发安全。事件处理器文档components/cqrs/event_handler.go明确指出同一 handler 实例可能被并发调用因此读模型必须线程安全活动时间线_examples/basic/6-cqrs-ordered-events/activity.go按到达顺序 appendActivityEntry时间戳、订阅者 ID、活动类型、详情并用结构化日志[ACTIVITY]输出每一条活动。数据契约protobuf 消息定义所有命令与事件都以 protobuf 定义_examples/basic/6-cqrs-ordered-events/proto/messages.protomessage MessageMetadata { string partition_key 1; google.protobuf.Timestamp created_at 2; } // Commands message Subscribe { MessageMetadata metadata 1; string subscriber_id 2; string email 3; } message Unsubscribe { MessageMetadata metadata 1; string subscriber_id 2; } message UpdateEmail { MessageMetadata metadata 1; string subscriber_id 2; string new_email 3; } // Events message SubscriberSubscribed { MessageMetadata metadata 1; string subscriber_id 2; string email 3; } message SubscriberUnsubscribed { MessageMetadata metadata 1; string subscriber_id 2; } message SubscriberEmailUpdated { MessageMetadata metadata 1; string subscriber_id 2; string new_email 3; }每个命令/事件都内嵌MessageMetadata携带partition_key订阅者 ID与created_at创建时间。重新生成 Go 代码的方式见 _examples/basic/6-cqrs-ordered-events/Makefileproto: protoc --proto_pathproto --go_out. --go_optpathssource_relative proto/messages.proto依赖版本可参考 _examples/basic/6-cqrs-ordered-events/go.modwatermill v1.5.1、watermill-kafka/v3 v3.1.2、google.golang.org/protobuf v1.36.11Go 1.25。运行示例前置条件Docker 与 Docker Compose示例目录自带 docker-compose.yml其中golang服务以go run .启动应用kafka服务使用 Redpandaredpandadata/redpanda:v26.1.7提供 Kafka 协议兼容的 broker并向宿主机暴露19092端口。启动步骤在示例目录下执行docker-compose up应用启动后会自动执行simulateTraffic_examples/basic/6-cqrs-ordered-events/main.go以每 500ms 一条的节奏循环产生真实流量生成新的subscriber_id发送Subscribe命令500ms 后发送UpdateEmail命令每 3 个周期额外发送一次Unsubscribe命令i%3 0。观察日志即可验证效果[ACTIVITY]前缀的时间线日志来自activity.go展示按序到达的活动流Subscriber added / updated / removed日志来自subscribers.go展示订阅者列表的状态变迁Received message调试日志展示每条消息的元数据可看到partition_key已写入。可优化方向拆分 topic 以水平扩展README 的 Possible improvements 一节给出了一个重要的扩展建议In this example, we are using globaleventsandcommandstopics. You can consider splitting them into smaller topics, for example, per aggregate type. Thanks to that, you can scale your application horizontally and increase the throughput and processing less events.即当前示例为了保序把所有命令和事件分别塞进全局唯一的commands/eventstopic这保证了全局顺序但限制了并行度。如果业务允许按聚合aggregate隔离顺序可以考虑按聚合类型拆分 topic——例如每个聚合一个 topic。这样不同聚合的 topic 可以各自独立扩展、并行消费吞吐量随之提升而每个聚合内部的顺序依然有保障。拆分方式非常直接把GeneratePublishTopic与GenerateSubscribeTopic从固定返回常量改为按命令/事件所属聚合类型返回 topic 名。例如基于cqrs.StructName按类型名生成 topic示例代码注释中也提到了这一点再配合分区键即可在聚合粒度上同时获得顺序与并行。小结这个示例浓缩了在 Watermill 中构建顺序敏感型CQRS 应用的三个核心实践**事件处理组EventGroupProcessor**让同一读模型的多个处理器共享一个订阅消息严格按到达顺序逐条处理**全局单一 topic Kafka 分区键subscriber_id**在维持每个订阅者事件顺序的同时允许不同订阅者的消息并行消费吞吐与顺序兼得Marshaler 装饰器把业务元数据分区键、创建时间安全地带入消息层让 Kafka 分区策略与业务键自然对齐。如需继续深入学习可以阅读仓库中 components/cqrs 组件的源码与测试例如 event_processor_group_test.go、event_processor_test.go以及 CQRS 文档 docs/content/docs/cqrs.md了解CommandBus、EventBus、Marshaler等的完整配置项与进阶用法。【免费下载链接】watermillBuilding event-driven applications the easy way in Go.项目地址: https://gitcode.com/GitHub_Trending/wa/watermill创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 使用 Watermill CQRS 与 Kafka 分区键实现严格有序事件处理：邮件订阅系统实战解析
使用 Watermill CQRS 与 Kafka 分区键实现严格有序事件处理邮件订阅系统实战解析【免费下载链接】watermillBuilding event-driven applications the easy way in Go.项目地址: https://gitcode.com/GitHub_Trending/wa/watermill本篇指南基于仓库中的完整可运行示例 _examples/basic/6-cqrs-ordered-eventsREADME 位于 _examples/basic/6-cqrs-ordered-events/README.md讲解如何在 Go 中基于 Watermill 的 CQRS 组件构建一个事件顺序敏感的邮件订阅系统用户可订阅、更新邮箱、退订系统同时维护活跃订阅者列表与订阅活动时间线两份读模型。读完后你将掌握Watermill CQRS 组件的基本装配方法、用**事件处理组Event Group**保证同一组事件消费有序的原理以及如何利用Kafka 分区键在不牺牲吞吐量的前提下维持每个订阅者事件的相对顺序。示例应用邮件订阅系统这是一个用 Watermill CQRS 组件实现的邮件订阅系统用户可以对邮件列表执行三种操作订阅提供邮箱地址加入订阅列表更新邮箱订阅后修改自己的邮箱地址退订从订阅列表中移除。系统始终维护两份视图Read Model当前活跃订阅者列表map[subscriberID]email实时反映谁是有效订阅者订阅活动时间线按时间顺序记录每一次SUBSCRIBED、UNSUBSCRIBED、EMAIL_UPDATED活动。系统整体遵循 CQRS 模式写操作通过Command命令发出由 Command Handler 处理后发布Event事件读模型则订阅事件流并增量更新。整个数据流由 message.Router 驱动这正是 Watermill 构建事件驱动应用的通用骨架。为什么事件顺序在这里是生死攸关的README 中明确指出了顺序的关键性In this example, keeping order of events is crucial. If events wont be ordered, andSubscriberSubscribedwould arrive afterSubscriberUnsubscribedevent, the subscriber will be still subscribed.保持事件顺序至关重要。如果事件乱序SubscriberSubscribed先于SubscriberUnsubscribed被处理那么用户明明退订了订阅者列表中却仍然保留着他。以活跃订阅者列表这个读模型为例它的状态变更完全依赖事件到达的顺序事件序列列表状态Subscribed→Unsubscribed用户被正确移除Unsubscribed→Subscribed乱序用户被错误地保留在列表中同理UpdateEmail与Unsubscribe之间也存在类似竞态若邮箱更新事件晚于退订事件到达一个已退订用户的新邮箱可能被重新写入列表。因此为每个订阅者的事件流维持严格顺序是本示例的核心工程问题而答案由两个机制共同给出事件处理组Event Group让多个事件处理器共享同一个订阅者实例保证同一读模型的事件按到达顺序逐个消费Kafka 分区键Partition Key以subscriber_id作为分区键让同一订阅者的所有消息落入 Kafka 的同一分区从而在不牺牲并行吞吐量的前提下维持每个订阅者的相对顺序。架构全景从总线到处理器示例在 _examples/basic/6-cqrs-ordered-events/main.go 中完成全部装配整体分为四层Kafka (commands topic) Kafka (events topic) │ │ ▼ ▼ CommandProcessor (3 个 Handler) EventGroupProcessor (2 个 Group) │ │ ▼ ▼ CommandBus ──► CommandHandler ──► EventBus ──► Read ModelCommandBus / EventBus负责发布命令与事件写入侧CommandProcessor订阅commandstopic按事件类型把命令分发给对应 HandlerEventGroupProcessor订阅eventstopic把同一读模型的一组事件处理器绑定到同一个 Kafka 订阅上。用事件处理组保证组内有序消费普通EventProcessor会为每一个EventHandler 单独创建一个订阅者参见 components/cqrs/event_processor.go 中SubscriberConstructor的注释而EventGroupProcessor则允许多个处理器共享同一个订阅者实例。这在 components/cqrs/event_processor_group.go 的注释中有明确说明Compared to EventProcessor, EventGroupProcessor allows to have multiple handlers that share the same subscriber instance.这带来两个直接好处单个订阅消费一个读模型对应一个 Kafka 消费组ConsumerGroup消息以严格顺序到达并被逐条分发组内路由当消息到达时Watermill 会根据消息中的事件名匹配组内的对应处理器见EventGroupProcessor.routerHandlerGroupFunccomponents/cqrs/event_processor_group.go未匹配到的事件会以 trace 日志跳过。示例中注册了两个事件组_examples/basic/6-cqrs-ordered-events/main.go// 组 1订阅者读模型共享同一个订阅 err eventProcessor.AddHandlersGroup( SubscriberReadModel, cqrs.NewGroupEventHandler(subscribersReadModel.OnSubscribed), cqrs.NewGroupEventHandler(subscribersReadModel.OnUnsubscribed), cqrs.NewGroupEventHandler(subscribersReadModel.OnEmailUpdated), ) // 组 2活动时间线读模型共享同一个订阅 err eventProcessor.AddHandlersGroup( ActivityTimelineReadModel, cqrs.NewGroupEventHandler(activityReadModel.OnSubscribed), cqrs.NewGroupEventHandler(activityReadModel.OnUnsubscribed), cqrs.NewGroupEventHandler(activityReadModel.OnEmailUpdated), )每个组在创建订阅者时都使用params.EventGroupName作为 Kafka 消费组名_examples/basic/6-cqrs-ordered-events/main.go即SubscriberReadModel、ActivityTimelineReadModel各自拥有独立的消费组互不干扰地各自按序消费。EventGroupProcessor 的核心配置项根据 components/cqrs/event_processor_group.goEventGroupProcessorConfig的关键字段如下配置项是否必填作用GenerateSubscribeTopic必填为事件组生成订阅 topic示例中统一返回eventsSubscriberConstructor必填为每个事件组创建订阅者每个组只调用一次这正是一组一订阅、有序消费的根基Marshaler必填事件/命令的编解码器OnHandle可选类似中间件可在事件处理前后注入逻辑须显式调用params.Handler.Handle()AckOnUnknownEvent可选事件无对应处理器时是否 ack默认不 ack 并报错Logger可选默认watermill.NopLogger单一 Topic 分区键顺序与吞吐兼得全局单一 topic 的意义为了让所有订阅者的事件流共享同一个有序消费序列命令与事件分别被路由到单一的全局 topic_examples/basic/6-cqrs-ordered-events/main.gocommandBus, err : cqrs.NewCommandBusWithConfig(publisher, cqrs.CommandBusConfig{ GeneratePublishTopic: func(params cqrs.CommandBusGeneratePublishTopicParams) (string, error) { // 所有命令共用一个 topic以保持命令的有序性 return commands, nil }, Marshaler: cqrsMarshaler, Logger: logger, }) eventBus, err : cqrs.NewEventBusWithConfig(publisher, cqrs.EventBusConfig{ GeneratePublishTopic: func(params cqrs.GenerateEventPublishTopicParams) (string, error) { // 所有事件共用一个 topic以保持事件的有序性 return events, nil }, Marshaler: cqrsMarshaler, Logger: logger, })这里用GeneratePublishTopic回调把 topic 固定为常量而不是按事件/命令类型生成对比cqrs.StructName按类型名命名 topic 的默认策略。为什么 Kafka 分区键是关键Kafka 保证同一分区内消息有序。如果所有消息都挤在一个分区虽然全局有序但吞吐量会被单分区上限卡死如果消息随机分散到多个分区顺序又无法保证。示例的解法是以subscriber_id作为分区键。同一订阅者的消息订阅、更新、退订永远落入同一分区Kafka 保证它们在分区内有序而不同订阅者的消息可以并行分布在不同分区由多个消费者并发处理——既保序又不损失吞吐。这正是 README 中使用 Kafka 分区键在保持事件顺序的同时提升处理吞吐的实现方式。在 _examples/basic/6-cqrs-ordered-events/main.go 中通过 watermill-kafka 提供的kafka.NewWithPartitioningMarshaler启用分区键kafkaMarshaler : kafka.NewWithPartitioningMarshaler(GenerateKafkaPartitionKey)GenerateKafkaPartitionKey从 Watermill 消息元数据中取出partition_key字段并返回_examples/basic/6-cqrs-ordered-events/message.gofunc GenerateKafkaPartitionKey(topic string, msg *message.Message) (string, error) { slog.Debug(Setting partition key, topic, topic, msg_metadata, msg.Metadata) return msg.Metadata.Get(PartitionKeyMetadataField), nil }分区键如何随消息流转分区键不会凭空产生示例通过装饰cqrs.ProtoMarshaler把 Proto 消息中的业务元数据搬运到 Watermill 消息元数据中_examples/basic/6-cqrs-ordered-events/message.gotype CqrsMarshalerDecorator struct { cqrs.ProtoMarshaler } const PartitionKeyMetadataField partition_key func (c CqrsMarshalerDecorator) Marshal(v interface{}) (*message.Message, error) { msg, err : c.ProtoMarshaler.Marshal(v) if err ! nil { return nil, err } pm, ok : v.(ProtoMessage) if !ok { return nil, fmt.Errorf(%T does not implement ProtoMessage and cant be marshaled, v) } metadata : pm.GetMetadata() if metadata nil { return nil, fmt.Errorf(%T.GetMetadata returned nil, v) } msg.Metadata.Set(PartitionKeyMetadataField, metadata.PartitionKey) msg.Metadata.Set(created_at, metadata.CreatedAt.AsTime().String()) return msg, nil }MessageMetadata是每个命令/事件的公共 Proto 字段_examples/basic/6-cqrs-ordered-events/proto/messages.proto由GenerateMessageMetadata(partitionKey string)统一创建_examples/basic/6-cqrs-ordered-events/message.go。在模拟流量中subscriber_id即分区键_examples/basic/6-cqrs-ordered-events/main.go。整条链路因此闭环simulateTraffic 生成命令 │ Metadata.PartitionKey subscriberID ▼ CqrsMarshalerDecorator 把 partition_key 写入消息元数据 ▼ kafka.NewWithPartitioningMarshaler 读取 partition_key 作为 Kafka 分区键 ▼ 同一订阅者的消息进入同一 Kafka 分区分区内有序 ▼ EventGroupProcessor 按序消费并分发到读模型处理器装配细节Router、中间件与处理器Router 与中间件CQRS 建立在 Watermill 消息路由之上components/cqrs 组件全部以*message.Router为基座。示例创建 Router 并挂载中间件_examples/basic/6-cqrs-ordered-events/main.gorouter, err : message.NewRouter(message.RouterConfig{}, logger) if err ! nil { panic(err) } // 恢复事件/命令处理器中的 panic防止消费者进程崩溃 router.AddMiddleware(middleware.Recoverer) // 自定义中间件记录每条收到的消息元数据 router.AddMiddleware(func(h message.HandlerFunc) message.HandlerFunc { return func(msg *message.Message) ([]*message.Message, error) { slog.Debug(Received message, metadata, msg.Metadata) return h(msg) } })middleware.Recoverer位于 message/router/middleware/recoverer.go是所有生产级 Watermill 应用的基础防护你还可以从 message/router/middleware 中选择重试Retry、超时Timeout、熔断CircuitBreaker、去重Deduplicator等中间件。Kafka 发布/订阅示例使用 Redpanda 提供 Kafka 兼容的 broker地址为kafka:9092。发布端_examples/basic/6-cqrs-ordered-events/main.gopublisher, err : kafka.NewPublisher( kafka.PublisherConfig{ Brokers: []string{kafka:9092}, Marshaler: kafkaMarshaler, // 带分区键的 marshaler }, watermillLogger, )订阅端由SubscriberConstructor按需创建命令处理器用params.HandlerName作为消费组名每个命令 handler 独立消费组事件组处理器用params.EventGroupName作为消费组名每个读模型一个消费组。命令处理器三个命令分别对应三个 handler_examples/basic/6-cqrs-ordered-events/main.goerr commandProcessor.AddHandlers( cqrs.NewCommandHandler(SubscribeHandler, SubscribeHandler{eventBus}.Handle), cqrs.NewCommandHandler(UnsubscribeHandler, UnsubscribeHandler{eventBus}.Handle), cqrs.NewCommandHandler(UpdateEmailHandler, UpdateEmailHandler{eventBus}.Handle), )每个 CommandHandler 的职责是把命令转换为领域事件并通过EventBus发布。例如SubscribeHandler_examples/basic/6-cqrs-ordered-events/main.gofunc (h SubscribeHandler) Handle(ctx context.Context, cmd *Subscribe) error { return h.eventBus.Publish(ctx, SubscriberSubscribed{ Metadata: GenerateMessageMetadata(cmd.SubscriberId), SubscriberId: cmd.SubscriberId, Email: cmd.Email, }) }UnsubscribeHandler与UpdateEmailHandler结构相同分别发布SubscriberUnsubscribed与SubscriberEmailUpdated。读模型纯内存 互斥锁两个读模型都定义在独立文件中订阅者列表_examples/basic/6-cqrs-ordered-events/subscribers.gomap[string]stringsubscriberID → emailOnSubscribed写入、OnUnsubscribed删除、OnEmailUpdated更新并通过sync.RWMutex保证并发安全。事件处理器文档components/cqrs/event_handler.go明确指出同一 handler 实例可能被并发调用因此读模型必须线程安全活动时间线_examples/basic/6-cqrs-ordered-events/activity.go按到达顺序 appendActivityEntry时间戳、订阅者 ID、活动类型、详情并用结构化日志[ACTIVITY]输出每一条活动。数据契约protobuf 消息定义所有命令与事件都以 protobuf 定义_examples/basic/6-cqrs-ordered-events/proto/messages.protomessage MessageMetadata { string partition_key 1; google.protobuf.Timestamp created_at 2; } // Commands message Subscribe { MessageMetadata metadata 1; string subscriber_id 2; string email 3; } message Unsubscribe { MessageMetadata metadata 1; string subscriber_id 2; } message UpdateEmail { MessageMetadata metadata 1; string subscriber_id 2; string new_email 3; } // Events message SubscriberSubscribed { MessageMetadata metadata 1; string subscriber_id 2; string email 3; } message SubscriberUnsubscribed { MessageMetadata metadata 1; string subscriber_id 2; } message SubscriberEmailUpdated { MessageMetadata metadata 1; string subscriber_id 2; string new_email 3; }每个命令/事件都内嵌MessageMetadata携带partition_key订阅者 ID与created_at创建时间。重新生成 Go 代码的方式见 _examples/basic/6-cqrs-ordered-events/Makefileproto: protoc --proto_pathproto --go_out. --go_optpathssource_relative proto/messages.proto依赖版本可参考 _examples/basic/6-cqrs-ordered-events/go.modwatermill v1.5.1、watermill-kafka/v3 v3.1.2、google.golang.org/protobuf v1.36.11Go 1.25。运行示例前置条件Docker 与 Docker Compose示例目录自带 docker-compose.yml其中golang服务以go run .启动应用kafka服务使用 Redpandaredpandadata/redpanda:v26.1.7提供 Kafka 协议兼容的 broker并向宿主机暴露19092端口。启动步骤在示例目录下执行docker-compose up应用启动后会自动执行simulateTraffic_examples/basic/6-cqrs-ordered-events/main.go以每 500ms 一条的节奏循环产生真实流量生成新的subscriber_id发送Subscribe命令500ms 后发送UpdateEmail命令每 3 个周期额外发送一次Unsubscribe命令i%3 0。观察日志即可验证效果[ACTIVITY]前缀的时间线日志来自activity.go展示按序到达的活动流Subscriber added / updated / removed日志来自subscribers.go展示订阅者列表的状态变迁Received message调试日志展示每条消息的元数据可看到partition_key已写入。可优化方向拆分 topic 以水平扩展README 的 Possible improvements 一节给出了一个重要的扩展建议In this example, we are using globaleventsandcommandstopics. You can consider splitting them into smaller topics, for example, per aggregate type. Thanks to that, you can scale your application horizontally and increase the throughput and processing less events.即当前示例为了保序把所有命令和事件分别塞进全局唯一的commands/eventstopic这保证了全局顺序但限制了并行度。如果业务允许按聚合aggregate隔离顺序可以考虑按聚合类型拆分 topic——例如每个聚合一个 topic。这样不同聚合的 topic 可以各自独立扩展、并行消费吞吐量随之提升而每个聚合内部的顺序依然有保障。拆分方式非常直接把GeneratePublishTopic与GenerateSubscribeTopic从固定返回常量改为按命令/事件所属聚合类型返回 topic 名。例如基于cqrs.StructName按类型名生成 topic示例代码注释中也提到了这一点再配合分区键即可在聚合粒度上同时获得顺序与并行。小结这个示例浓缩了在 Watermill 中构建顺序敏感型CQRS 应用的三个核心实践**事件处理组EventGroupProcessor**让同一读模型的多个处理器共享一个订阅消息严格按到达顺序逐条处理**全局单一 topic Kafka 分区键subscriber_id**在维持每个订阅者事件顺序的同时允许不同订阅者的消息并行消费吞吐与顺序兼得Marshaler 装饰器把业务元数据分区键、创建时间安全地带入消息层让 Kafka 分区策略与业务键自然对齐。如需继续深入学习可以阅读仓库中 components/cqrs 组件的源码与测试例如 event_processor_group_test.go、event_processor_test.go以及 CQRS 文档 docs/content/docs/cqrs.md了解CommandBus、EventBus、Marshaler等的完整配置项与进阶用法。【免费下载链接】watermillBuilding event-driven applications the easy way in Go.项目地址: https://gitcode.com/GitHub_Trending/wa/watermill创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考