国际物流和3PL仓储业务，asp.net core web api 系统，为了解决在一个业务方法内直接调用多个系统的接口，数据可能不一致（当某个接口调用失败时,导致没有通知到该系统），
还有接口很慢的问题（内部调用的第三方接口有点多），引入了CAP和Kafka。
保存业务数据时，插入多条message，对应多个topic：
1）CAP.Published 表会插入 4 条记录
2）每个系统订阅自己专属的 Topic
3）适合不同系统需要不同的消息结构或字段
1 public async Task<bool> CreateFreightOrderHandle(CreateOrderCommand request, CancellationToken cancellationToken) 2 { 3 var order = new Order(...); 4 await _orderRepository.AddAsync(order); 5 6 // 发布 4 条消息到不同 Topic 7 await _capPublisher.PublishAsync("order.created.v1.warehouse", new OrderCreatedEvent(...)); 8 await _capPublisher.PublishAsync("order.created.v1.billing", new OrderCreatedEvent(...)); 9 await _capPublisher.PublishAsync("order.created.v1.tracking", new OrderCreatedEvent(...)); 10 await _capPublisher.PublishAsync("order.created.v1.notification", new OrderCreatedEvent(...)); 11 12 // 一次性提交：订单数据 + 4 条 Outbox 记录 13 await _orderRepository.UnitOfWork.SaveChangesAsync(cancellationToken); 14 15 return true; 16 }
1 public async Task<bool> CreateFreightOrderHandle(CreateOrderCommand request, CancellationToken cancellationToken) 2 { 3 var order = new Order(...); 4 await _orderRepository.AddAsync(order); 5 6 // 只发布 1 条消息 7 await _capPublisher.PublishAsync("order.created.v1", new OrderCreatedEvent(...)); 8 9 await _orderRepository.UnitOfWork.SaveChangesAsync(cancellationToken); 10 11 return true; 12 }
消费方：
// 仓库服务 [CapSubscribe("order.created.v1", Group = "warehouse.service")] public async Task HandleWarehouse(OrderCreatedEvent @event) { ... } // 计费服务 [CapSubscribe("order.created.v1", Group = "billing.service")] public async Task HandleBilling(OrderCreatedEvent @event) { ... } // 物流跟踪服务 [CapSubscribe("order.created.v1", Group = "tracking.service")] public async Task HandleTracking(OrderCreatedEvent @event) { ... } // 通知服务 [CapSubscribe("order.created.v1", Group = "notification.service")] public async Task HandleNotification(OrderCreatedEvent @event) { ... }
单topic特点：
1）CAP.Published 表只插入 1 条记录
2）Kafka 会自动把这条消息分发给 4 个 Consumer Group
3）减少数据库写入压力，更符合消息队列的设计理念
混合方案，解决特殊场景消息数据结构不一致问题
适合：系统正在快速迭代，业务变化频繁
// 核心事件用单 Topic + 向后兼容 await _capPublisher.PublishAsync("order.created.v1", new OrderCreatedEvent { OrderId = order.Id, Amount = order.Amount, // 只放核心通用字段 }); // 特殊需求用专属 Topic if (order.NeedCustomsClearance) { await _capPublisher.PublishAsync("order.customs.required.v1", new CustomsEvent { OrderId = order.Id, ClearanceCode = order.CustomsClearanceCode, Documents = order.CustomsDocuments }); }
// 发布端同时发两个版本 await _capPublisher.PublishAsync("order.created.v1", eventV1); // 老系统继续用 await _capPublisher.PublishAsync("order.created.v2", eventV2); // 新系统用新版 // 过渡期结束后，逐步下线 v1
优点：
1）完全隔离新老系统，零影响
2）适合多团队独立迭代
缺点：
1）过渡期需要维护两套消息（Outbox 表多一倍记录）
2）需要明确的下线时间表
注：文章内容由AI生成，仅做个人记录；
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。