后端技术17-Kafka vs RabbitMQ：事件驱动架构的消息队列选型

「知识图谱生成工具」一键将文件夹内容变身为交互式知识图谱的免安装桌面工具文末附免费下载链接-CSDN博客AI工程师面试高频考点问题汇总下载链接你是否经历过这样的噩梦凌晨3点支付系统挂了订单系统跟着崩库存系统也乱了套整个链路像多米诺骨牌一样全倒别慌今天咱们聊聊怎么用事件驱动架构(EDA)把系统从一损俱损变成各自安好。目录一、为什么需要事件驱动架构二、EDA核心概念从打电话到发微信三、消息队列选型Kafka、RabbitMQ、RocketMQ怎么选四、事件溯源(Event Sourcing)时光倒流的黑科技五、CQRS模式读写分离的艺术六、实战电商订单系统事件驱动改造七、数据一致性保障Saga模式八、总结与展望一、为什么需要事件驱动架构先讲个真实的故事。去年我接手一个电商系统架构是这样的┌─────────┐ HTTP ┌─────────┐ HTTP ┌─────────┐ │ 订单 │ ────────→ │ 库存 │ ────────→ │ 支付 │ │ 服务 │ │ 服务 │ │ 服务 │ └─────────┘ └─────────┘ └─────────┘看起来挺清晰对吧直到有一天库存服务响应慢了3秒订单服务的线程池被打满接着支付服务也开始超时最后整个系统雪崩。这就是同步调用的痛点服务间强耦合一个慢全都慢级联故障一崩全崩扩容必须整条链路一起扩高峰期流量直接压垮下游而事件驱动架构的核心思想很简单别打电话了发微信吧。┌─────────┐ 发布事件 ┌─────────┐ 订阅事件 ┌─────────┐ │ 订单 │ ─────────→ │ 消息 │ ←───────── │ 库存 │ │ 服务 │ │ 队列 │ │ 服务 │ └─────────┘ └─────────┘ └─────────┘ ↑ │ 订阅事件 ┌─────────┐ │ 支付 │ │ 服务 │ └─────────┘订单服务只负责说订单创建了至于库存扣不扣、支付怎么处理它不关心。各服务通过消息队列异步通信解耦、削峰、容错一举三得。二、EDA核心概念从打电话到发微信2.1 事件(Event)事件是已经发生的事情不可变、只增不减。public class OrderCreatedEvent { private String orderId; private String userId; private BigDecimal amount; private ListOrderItem items; private long timestamp; // ... getter/setter }关键特征不可变事件一旦发生不能修改可以补偿但不能改原事件时序性每个事件都有时间戳和版本号自包含事件携带完整上下文消费者不需要再查数据库2.2 发布-订阅模式┌─────────────────────────────────────────────────────────┐ │ 消息队列 (Broker) │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ order.topic │ │ pay.topic │ │ stock.topic │ │ │ │ (订单主题) │ │ (支付主题) │ │ (库存主题) │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ └─────────┼──────────────────┼──────────────────┼─────────┘ │ │ │ ┌─────┴─────┐ ┌─────┴─────┐ ┌─────┴─────┐ │ 订单服务 │ │ 支付服务 │ │ 库存服务 │ │ (发布者) │ │ (订阅者) │ │ (订阅者) │ └───────────┘ └───────────┘ └───────────┘发布者只管发订阅者只管收中间靠消息队列解耦。2.3 事件驱动 vs 请求响应特性请求响应(同步)事件驱动(异步)耦合度高直接依赖低通过消息队列响应时间实时最终一致容错性低级联故障高失败可重试扩展性需协调扩容独立扩容复杂度简单直接需要处理幂等、顺序三、消息队列选型Kafka、RabbitMQ、RocketMQ怎么选选消息队列就像选对象没有最好的只有最合适的。3.1 三款MQ对比┌─────────────────┬─────────────┬─────────────┬─────────────┐ │ 特性 │ Kafka │ RabbitMQ │ RocketMQ │ ├─────────────────┼─────────────┼─────────────┼─────────────┤ │ 开发语言 │ Scala │ Erlang │ Java │ │ 吞吐量 │ 百万级 │ 万级 │ 十万级 │ │ 延迟 │ 毫秒级 │ 微秒级 │ 毫秒级 │ │ 消息持久化 │ ✓ │ ✓ │ ✓ │ │ 事务消息 │ ✗ │ ✓ │ ✓ │ │ 顺序消息 │ ✓ │ ✗ │ ✓ │ │ 延迟消息 │ ✗ │ ✓ │ ✓ │ │ 死信队列 │ ✗ │ ✓ │ ✓ │ │ 社区活跃度 │ 高 │ 高 │ 中 │ └─────────────────┴─────────────┴─────────────┴─────────────┘3.2 场景推荐选Kafka日志采集、大数据处理需要超高吞吐量消息顺序性要求高单分区需要流处理Kafka Streams选RabbitMQ需要复杂路由规则延迟要求极低微秒级需要灵活的消息模式RPC、工作队列等团队Erlang经验丰富选RocketMQ金融级可靠性要求需要事务消息需要延迟消息、定时消息国内生态文档中文友好3.3 代码示例Spring Boot集成Kafka// 生产者配置 Configuration public class KafkaProducerConfig { Bean public ProducerFactoryString, Object producerFactory() { MapString, Object config new HashMap(); config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092); config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); // 开启幂等性自动处理消息去重 config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); return new DefaultKafkaProducerFactory(config); } Bean public KafkaTemplateString, Object kafkaTemplate() { return new KafkaTemplate(producerFactory()); } } // 事件发布 Service public class OrderEventPublisher { Autowired private KafkaTemplateString, Object kafkaTemplate; public void publishOrderCreated(Order order) { OrderCreatedEvent event new OrderCreatedEvent( order.getId(), order.getUserId(), order.getAmount(), order.getItems(), System.currentTimeMillis() ); // 发送消息key用订单ID保证同一订单的顺序性 kafkaTemplate.send(order-created-topic, order.getId(), event) .whenComplete((result, ex) - { if (ex ! null) { // 记录日志进入补偿流程 log.error(发送订单创建事件失败, ex); } }); } } // 消费者配置 Configuration public class KafkaConsumerConfig { Bean public ConsumerFactoryString, Object consumerFactory() { MapString, Object config new HashMap(); config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092); config.put(ConsumerConfig.GROUP_ID_CONFIG, inventory-service); config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class); // 手动提交偏移量确保业务处理完再确认 config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); return new DefaultKafkaConsumerFactory(config); } Bean public ConcurrentKafkaListenerContainerFactoryString, Object kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactoryString, Object factory new ConcurrentKafkaListenerContainerFactory(); factory.setConsumerFactory(consumerFactory()); return factory; } } // 事件监听 Component Slf4j public class OrderEventListener { Autowired private InventoryService inventoryService; KafkaListener(topics order-created-topic, groupId inventory-service) public void onOrderCreated(Payload OrderCreatedEvent event, Acknowledgment acknowledgment) { try { log.info(收到订单创建事件: {}, event.getOrderId()); // 扣减库存 inventoryService.deductStock(event.getItems()); // 业务处理成功确认消息 acknowledgment.acknowledge(); } catch (Exception e) { log.error(处理订单创建事件失败: {}, event.getOrderId(), e); // 不确认消息会重新投递 throw e; } } }四、事件溯源(Event Sourcing)时光倒流的黑科技传统CRUD是这样的┌─────────┐ ┌─────────┐ │ 应用 │ ←──→ │ 数据库 │ │ 状态 │ 直接更新 │ 当前状态 │ └─────────┘ └─────────┘事件溯源是这样的┌─────────┐ 产生事件 ┌─────────┐ 存储事件 ┌─────────┐ │ 应用 │ ─────────→ │ 事件 │ ─────────→ │ 事件 │ │ 操作 │ │ 对象 │ │ 存储 │ └─────────┘ └─────────┘ └────┬────┘ │ ┌─────────────────────────┘ │ 重放事件 ↓ ┌─────────────┐ │ 还原当前状态 │ └─────────────┘4.1 核心思想不存最终状态只存事件流。当前状态 所有事件重放的结果。// 传统方式 public class Account { private String id; private BigDecimal balance; // 只存当前余额 } // 事件溯源方式 public class Account { private String id; private ListDomainEvent events new ArrayList(); // 存所有事件 // 通过事件重放计算当前余额 public BigDecimal getBalance() { return events.stream() .filter(e - e instanceof MoneyDeposited || e instanceof MoneyWithdrawn) .map(e - { if (e instanceof MoneyDeposited) return ((MoneyDeposited) e).getAmount(); else return ((MoneyWithdrawn) e).getAmount().negate(); }) .reduce(BigDecimal.ZERO, BigDecimal::add); } }4.2 事件存储示例事件ID聚合根ID事件类型事件数据版本号时间戳evt_001acc_123AccountCreated{“owner”:“张三”}12024-01-01 10:00:00evt_002acc_123MoneyDeposited{“amount”:1000}22024-01-02 14:30:00evt_003acc_123MoneyWithdrawn{“amount”:200}32024-01-03 09:15:00当前余额 1000 - 200 8004.3 事件溯源的优势完整的审计日志知道每一笔变化的来龙去脉时间旅行可以回到任意时间点的状态调试神器事件流就是复现问题的剧本灵活建模可以随时增加新的投影视图4.4 事件溯源的坑事件 schema 变更老事件怎么兼容新代码性能问题事件太多重放慢怎么办复杂度团队理解成本高查询困难只能通过事件ID查复杂查询需要投影解决方案CQRS 快照五、CQRS模式读写分离的艺术CQRS(Command Query Responsibility Segregation)把读写彻底分开┌─────────────────────────────────────────────────────────────────┐ │ 客户端 │ └──────────────────────────────┬──────────────────────────────────┘ │ ┌───────────────────┴───────────────────┐ │ │ ↓ ↓ ┌─────────────────────┐ ┌─────────────────────┐ │ 命令端 (C) │ │ 查询端 (Q) │ │ 处理写操作 │ │ 处理读操作 │ │ 生成事件 │ │ 返回视图 │ └──────────┬──────────┘ └──────────┬──────────┘ │ │ ↓ ↓ ┌─────────────────────┐ ┌─────────────────────┐ │ 事件存储 │ │ 读模型/投影 │ │ (Event Store) │ ──同步事件──→ │ (Read Model) │ │ 只追加事件 │ │ 优化查询的数据库 │ └─────────────────────┘ └─────────────────────┘5.1 代码示例// 命令端 RestController RequestMapping(/commands/orders) public class OrderCommandController { Autowired private OrderCommandService commandService; PostMapping public ResponseEntityString createOrder(RequestBody CreateOrderCommand command) { String orderId commandService.createOrder(command); return ResponseEntity.ok(orderId); } PostMapping(/{orderId}/pay) public ResponseEntityVoid payOrder(PathVariable String orderId) { commandService.payOrder(new PayOrderCommand(orderId)); return ResponseEntity.ok().build(); } } Service public class OrderCommandService { Autowired private EventSourcingRepositoryOrderAggregate repository; public String createOrder(CreateOrderCommand command) { OrderAggregate order new OrderAggregate(); order.create(command.getUserId(), command.getItems(), command.getAddress()); repository.save(order); return order.getId(); } public void payOrder(PayOrderCommand command) { OrderAggregate order repository.findById(command.getOrderId()); order.pay(command.getPaymentInfo()); repository.save(order); } } // 查询端 RestController RequestMapping(/queries/orders) public class OrderQueryController { Autowired private OrderQueryService queryService; GetMapping(/{orderId}) public OrderView getOrder(PathVariable String orderId) { return queryService.getOrder(orderId); } GetMapping(/user/{userId}) public ListOrderSummaryView getUserOrders(PathVariable String userId) { return queryService.getUserOrders(userId); } } Service public class OrderQueryService { Autowired private OrderReadRepository readRepository; public OrderView getOrder(String orderId) { // 直接从读模型查询可以JOIN、可以缓存 return readRepository.findById(orderId); } public ListOrderSummaryView getUserOrders(String userId) { // 针对查询优化的数据结构 return readRepository.findByUserIdOrderByCreateTimeDesc(userId); } } // 事件处理器同步命令端到查询端 Component public class OrderEventHandler { Autowired private OrderReadRepository readRepository; EventListener public void on(OrderCreatedEvent event) { OrderView view new OrderView(); view.setOrderId(event.getOrderId()); view.setUserId(event.getUserId()); view.setStatus(CREATED); view.setCreateTime(event.getTimestamp()); readRepository.save(view); } EventListener public void on(OrderPaidEvent event) { OrderView view readRepository.findById(event.getOrderId()); view.setStatus(PAID); view.setPayTime(event.getTimestamp()); readRepository.save(view); } }5.2 CQRS的优势独立优化读模型和写模型可以分别优化团队协作前端只管查询后端专注领域逻辑性能提升读模型可以大量缓存、反规范化安全性命令端可以严格校验查询端只读六、实战电商订单系统事件驱动改造6.1 改造前的同步架构用户下单 → 订单服务 → 调用库存服务 → 调用支付服务 → 返回结果 ↑ └──────── 任一环节失败全部回滚 ────────┘问题库存服务慢订单服务跟着慢支付失败需要同步回滚库存高峰期支付服务被打挂6.2 改造后的事件驱动架构┌─────────────────────────────────────────────────────────────────┐ │ 订单服务 (Order Service) │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ 命令端createOrder() → 发布 OrderCreatedEvent │ │ │ │ payOrder() → 发布 OrderPaidEvent │ │ │ └─────────────────────────────────────────────────────────┘ │ └──────────────────────────────────┬────────────────────────────┘ │ 发布事件 ↓ ┌─────────────────────────────────────────────────────────────────┐ │ 消息队列 (Kafka) │ │ ┌─────────────────┬─────────────────┬─────────────────┐ │ │ │ order-created │ order-paid │ stock-deducted │ │ │ │ topic │ topic │ topic │ │ │ └────────┬────────┴────────┬────────┴────────┬────────┘ │ └───────────┼─────────────────┼─────────────────┼────────────────┘ │ │ │ ┌────────┘ │ └────────┐ ↓ ↓ ↓ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 库存服务 │ │ 支付服务 │ │ 物流服务 │ │ │ │ │ │ │ │ 订阅订单创建 │ │ 订阅订单创建 │ │ 订阅支付完成 │ │ 扣减库存 │ │ 创建支付单 │ │ 创建物流单 │ │ 发布库存扣减 │ │ 发布支付完成 │ │ │ │ 事件 │ │ 事件 │ │ │ └─────────────┘ └─────────────┘ └─────────────┘6.3 核心事件定义// 订单创建事件 Data public class OrderCreatedEvent implements DomainEvent { private String eventId UUID.randomUUID().toString(); private String orderId; private String userId; private BigDecimal totalAmount; private ListOrderItem items; private ShippingAddress address; private long timestamp System.currentTimeMillis(); private int version 1; } // 库存扣减事件 Data public class StockDeductedEvent implements DomainEvent { private String eventId UUID.randomUUID().toString(); private String orderId; private ListStockItem deductedItems; private long timestamp System.currentTimeMillis(); private int version 1; } // 支付完成事件 Data public class OrderPaidEvent implements DomainEvent { private String eventId UUID.randomUUID().toString(); private String orderId; private String paymentId; private BigDecimal paidAmount; private PaymentMethod paymentMethod; private long timestamp System.currentTimeMillis(); private int version 1; }6.4 订单服务核心代码Service Slf4j public class OrderService { Autowired private OrderRepository orderRepository; Autowired private EventPublisher eventPublisher; /** * 创建订单 - 只负责发布事件不直接调用其他服务 */ Transactional public Order createOrder(CreateOrderRequest request) { // 1. 创建订单聚合根 Order order Order.create( IdGenerator.nextId(), request.getUserId(), request.getItems(), request.getAddress() ); // 2. 保存订单 orderRepository.save(order); // 3. 发布订单创建事件 OrderCreatedEvent event OrderCreatedEvent.from(order); eventPublisher.publish(order-created-topic, event); log.info(订单创建成功已发布事件: {}, order.getId()); return order; } /** * 处理支付回调 - 发布支付完成事件 */ Transactional public void handlePaymentCallback(String orderId, PaymentResult result) { Order order orderRepository.findById(orderId) .orElseThrow(() - new OrderNotFoundException(orderId)); // 1. 更新订单状态 order.markAsPaid(result.getPaymentId(), result.getPaidAmount()); orderRepository.save(order); // 2. 发布支付完成事件 OrderPaidEvent event OrderPaidEvent.from(order, result); eventPublisher.publish(order-paid-topic, event); log.info(订单支付完成已发布事件: {}, orderId); } }6.5 库存服务事件监听Component Slf4j public class InventoryEventListener { Autowired private InventoryService inventoryService; Autowired private EventPublisher eventPublisher; /** * 监听订单创建事件扣减库存 */ KafkaListener(topics order-created-topic, groupId inventory-service) public void handleOrderCreated(Payload OrderCreatedEvent event) { log.info(收到订单创建事件开始扣减库存: {}, event.getOrderId()); try { // 1. 扣减库存 ListStockItem deductedItems inventoryService.deductStock(event.getItems()); // 2. 发布库存扣减成功事件 StockDeductedEvent deductedEvent new StockDeductedEvent( event.getOrderId(), deductedItems ); eventPublisher.publish(stock-deducted-topic, deductedEvent); log.info(库存扣减成功: {}, event.getOrderId()); } catch (InsufficientStockException e) { // 3. 库存不足发布库存扣减失败事件 StockDeductionFailedEvent failedEvent new StockDeductionFailedEvent( event.getOrderId(), e.getSkuId(), e.getRequiredQty(), e.getAvailableQty() ); eventPublisher.publish(stock-deduction-failed-topic, failedEvent); log.warn(库存不足订单: {}, SKU: {}, event.getOrderId(), e.getSkuId()); } } }七、数据一致性保障Saga模式事件驱动架构最大的挑战分布式事务怎么保证7.1 Saga模式原理Saga把一个长事务拆成多个本地事务每个本地事务有对应的补偿操作。正常流程 T1(创建订单) → T2(扣减库存) → T3(扣款) → T4(发货) T3失败时的补偿流程 T1(创建订单) → T2(扣减库存) → T3(扣款) ✗ 失败 ↓ C3(退款) → C2(恢复库存) → C1(取消订单)7.2 Saga两种实现方式编排式(Choreography)各服务监听事件自主决定下一步订单服务 ──OrderCreated──→ 库存服务 ──StockDeducted──→ 支付服务 ↑ │ └────────────OrderPaid←─────────────────────────┘协调式(Orchestration)由Saga协调器统一调度┌─────────────┐ │ Saga协调器 │──→ 调用订单服务创建订单 │ │←── 成功 │ │──→ 调用库存服务扣减库存 │ │←── 成功 │ │──→ 调用支付服务扣款 │ │←── 失败 │ │──→ 调用库存服务恢复库存 │ │──→ 调用订单服务取消订单 └─────────────┘7.3 代码示例编排式SagaComponent public class OrderSagaOrchestrator { Autowired private SagaStateRepository sagaRepository; Autowired private EventPublisher eventPublisher; /** * 启动Saga流程 */ public void startSaga(CreateOrderRequest request) { String sagaId UUID.randomUUID().toString(); SagaState saga new SagaState(); saga.setSagaId(sagaId); saga.setStatus(SagaStatus.STARTED); saga.setSteps(Arrays.asList( new SagaStep(CREATE_ORDER, SagaStepStatus.PENDING), new SagaStep(DEDUCT_STOCK, SagaStepStatus.PENDING), new SagaStep(PROCESS_PAYMENT, SagaStepStatus.PENDING) )); sagaRepository.save(saga); // 第一步创建订单 eventPublisher.publish(saga-create-order, new SagaCreateOrderCommand(sagaId, request)); } /** * 处理步骤完成事件 */ EventListener public void onStepCompleted(SagaStepCompletedEvent event) { SagaState saga sagaRepository.findById(event.getSagaId()); saga.markStepCompleted(event.getStepName()); // 根据当前状态决定下一步 switch (event.getStepName()) { case CREATE_ORDER: // 订单创建成功扣减库存 eventPublisher.publish(saga-deduct-stock, new SagaDeductStockCommand(saga.getSagaId(), saga.getOrderId())); break; case DEDUCT_STOCK: // 库存扣减成功处理支付 eventPublisher.publish(saga-process-payment, new SagaProcessPaymentCommand(saga.getSagaId(), saga.getOrderId())); break; case PROCESS_PAYMENT: // 支付成功Saga完成 saga.setStatus(SagaStatus.COMPLETED); break; } sagaRepository.save(saga); } /** * 处理步骤失败事件触发补偿 */ EventListener public void onStepFailed(SagaStepFailedEvent event) { SagaState saga sagaRepository.findById(event.getSagaId()); saga.setStatus(SagaStatus.COMPENSATING); // 逆向执行补偿操作 ListSagaStep completedSteps saga.getCompletedSteps(); for (int i completedSteps.size() - 1; i 0; i--) { SagaStep step completedSteps.get(i); publishCompensationEvent(step.getName(), saga); } sagaRepository.save(saga); } private void publishCompensationEvent(String stepName, SagaState saga) { switch (stepName) { case CREATE_ORDER: eventPublisher.publish(saga-cancel-order, new SagaCancelOrderCommand(saga.getSagaId(), saga.getOrderId())); break; case DEDUCT_STOCK: eventPublisher.publish(saga-restore-stock, new SagaRestoreStockCommand(saga.getSagaId(), saga.getOrderId())); break; case PROCESS_PAYMENT: eventPublisher.publish(saga-refund-payment, new SagaRefundPaymentCommand(saga.getSagaId(), saga.getOrderId())); break; } } }7.4 Saga设计要点幂等性每个操作都要保证幂等防止重复执行可观测性Saga状态要持久化方便排查问题补偿顺序按正向操作的逆序执行补偿补偿也可能失败需要重试或人工介入八、总结与展望本文要点回顾事件驱动架构把同步调用变成异步消息解耦服务、削峰填谷消息队列选型Kafka适合大数据RabbitMQ适合复杂路由RocketMQ适合金融场景事件溯源存事件而不是存状态获得完整的审计能力和时间旅行能力CQRS读写分离独立优化查询性能大幅提升Saga模式分布式事务的解决方案通过补偿保证最终一致性源码获取本文完整代码已上传GitHub https://github.com/example/eda-demo包含Spring Boot Kafka完整示例事件溯源框架实现Saga协调器代码Docker Compose一键启动环境思考题你的系统有哪些场景适合改造成事件驱动架构如果消息队列挂了你的系统能优雅降级吗事件溯源的事件数量爆炸式增长你会怎么优化系列预告下一篇《从0到1搭建高可用Kafka集群》再下一篇《事件驱动架构下的监控与告警实战》投票时间你在用消息队列吗用的哪个A. Kafka大数据场景首选B. RabbitMQ灵活路由好用C. RocketMQ阿里出品必属精品D. 还没用正在观望评论区告诉我你的选择作者简介10年Java后端开发经历过电商、金融、物流多个行业从CRUD boy成长为架构师。喜欢把复杂的技术讲简单如果你觉得本文有帮助欢迎点赞收藏转发三连标签事件驱动, EDA, Kafka, 消息队列, CQRS, 微服务, 架构设计