事件驱动架构EDA与事件中心完整知识梳理

事件驱动架构EDA与事件中心完整知识梳理

一、什么是"事件"

概念

事件（Event）是系统中发生的一个"事实"的描述。它表示"某件事已经发生了"，而不是"请你做某件事"。

事件的三要素

{"type":"order.created",// 事件类型：发生了什么"source":"order-service",// 事件来源：谁产生的"data":{"orderId":"123"},// 事件数据：具体内容"time":"2026-07-14T10:28:00Z"// 事件时间：什么时候发生的}

注：

博客：

https://blog.csdn.net/badao_liumang_qizhi

二、事件驱动架构（EDA）

核心模型

┌──────────┐ ┌──────────────┐ ┌──────────┐ │ 生产者 │ 发布事件 │ 事件中心 │ 推送事件 │ 消费者 │ │ Producer │ ────────→ │ Event Broker │ ────────→ │ Consumer │ └──────────┘ └──────────────┘ └──────────┘ │ │ 推送事件 ▼ ┌──────────┐ │ 消费者 B │ └──────────┘

与传统同步调用对比

同步调用（点对点）：

a服务 ──HTTP POST──→ b服务 ←──Response────

问题：b 挂了，a 的业务也跟着失败。

事件驱动（发布/订阅）：

a服务 ──发布事件──→ 事件中心 ──推送──→ b服务 ──推送──→ 监控服务 ──推送──→ 日志服务

优势：b挂了不影响 a；将来新增消费者无需改 a 代码。

适用场景

• 系统间异步通知（不需要同步等待结果）
• 一个事件需要通知多个下游系统
• 生产者和消费者由不同团队维护
• 需要解耦上下游系统的部署和发布周期

三、CloudEvents 规范

CloudEvents 是 CNCF（云原生计算基金会）定义的事件格式标准，目的是让不同系统产生的事件有统一的格式。

核心属性

HTTP 协议绑定

CloudEvents 通过 HTTP Header 传递元信息：

POST /event-center/api/v3/eventTypes/aaa.xxx/send Content-Type: application/json Ce-Type: aaa.xxx Ce-Specversion: 1.0 Authorization: Bearer <token> { "fulfillmentCode": "xx.20260714.000005.xx", "event": "ttt", "eventDate": "2026-07-14 10:28:00", "systemName": "uuu" }

四、事件中心的角色

事件中心（Event Center/Event Broker）是事件流转的中间枢纽：

┌─────────────────────────────────────────────────────────┐ │ 事件中心 │ │ │ │ ┌────────────┐ ┌────────────┐ ┌────────────────┐ │ │ │ 事件接收 │ │ 事件路由 │ │ 事件投递 │ │ │ │ (Ingress) │→ │ (Routing) │→ │ (Delivery) │ │ │ └────────────┘ └────────────┘ └────────────────┘ │ │ │ │ 职责： │ │ 1. 接收生产者发布的事件 │ │ 2. 根据 eventType 路由到对应订阅者 │ │ 3. 可靠投递（重试、死信队列） │ │ 4. 认证鉴权 │ └─────────────────────────────────────────────────────────┘

与 MQ 的区别

五、通用代码示例

1. 事件发布工具类

/** * 事件中心推送工具类. * 基于 CloudEvents 规范，通过 HTTP POST 推送事件到事件中心. */@Component@Slf4jpublicclassEventPublisher{@Value("${event-center.token}")privateStringtoken;privatefinalRestTemplaterestTemplate;publicEventPublisher(RestTemplaterestTemplate){this.restTemplate=restTemplate;}/** * 发布事件到事件中心. * * @param eventType 事件类型，如 "order.created" * @param url 事件中心接收地址 * @param payload 事件数据（JSON 字符串） * @return 推送结果 */publicEventResultpublish(StringeventType,Stringurl,Stringpayload){try{HttpHeadersheaders=newHttpHeaders();headers.setContentType(MediaType.APPLICATION_JSON);headers.set("Authorization",token);headers.set("Ce-Type",eventType);headers.set("Ce-Specversion","1.0");HttpEntity<String>request=newHttpEntity<>(payload,headers);ResponseEntity<String>response=restTemplate.postForEntity(url,request,String.class);log.info("事件发布成功, type={}, response={}",eventType,response.getBody());returnEventResult.success(response.getBody());}catch(Exceptione){log.warn("事件发布失败, type={}",eventType,e);returnEventResult.fail(e.getMessage());}}}

2. 本地消息表 + 可靠发布

/** * 可靠事件发布服务. * 先写本地消息表，事务提交后再触发实际推送. */@ServicepublicclassReliableEventPublisher{@ResourceprivateEventLogRepositoryeventLogRepository;@ResourceprivateEventMqSendereventMqSender;/** * 记录待发布事件（在业务事务内调用）. */@TransactionalpublicvoidprepareEvent(StringeventType,Stringpayload,StringbizKey){// 1. 写入本地消息表EventLogeventLog=newEventLog();eventLog.setEventType(eventType);eventLog.setPayload(payload);eventLog.setBizKey(bizKey);eventLog.setStatus(0);// 待发送eventLog.setRetryCount(0);eventLogRepository.save(eventLog);// 2. 注册事务提交后的回调，发送 MQAfterTransactionActionCollectorcollector=newAfterTransactionActionCollector();TransactionSynchronizationManager.registerSynchronization(collector);collector.addCommitSyncAction(()->eventMqSender.send(eventLog.getId()));}}

3. MQ 消费者 + 实际推送

/** * 事件推送 MQ 消费者. * 消费本地消息表 ID，执行实际推送. */@Component@Slf4jpublicclassEventPublishConsumer{@ResourceprivateEventLogRepositoryeventLogRepository;@ResourceprivateEventPublishereventPublisher;@RabbitListener(queues="${event.publish.queue}")publicvoidconsume(IntegereventLogId){EventLogeventLog=eventLogRepository.findById(eventLogId).orElse(null);if(eventLog==null){return;}// 已成功的跳过if(eventLog.getStatus()==1){return;}// 推送事件中心Stringurl=resolveUrl(eventLog.getEventType());EventResultresult=eventPublisher.publish(eventLog.getEventType(),url,eventLog.getPayload());// 更新状态if(result.isSuccess()){eventLog.setStatus(1);eventLog.setResponse(result.getBody());}else{eventLog.setStatus(2);eventLog.setErrorMsg(result.getErrorMsg());}eventLog.setRetryCount(eventLog.getRetryCount()+1);eventLogRepository.save(eventLog);}}

4. 事务同步回调机制

/** * Spring 事务同步回调示例. * 确保 MQ 消息在事务提交后才发送. */publicclassTransactionCallbackDemo{publicvoidbusinessMethod(){// ... 业务逻辑，操作数据库 ...// 注册事务提交后的回调TransactionSynchronizationManager.registerSynchronization(newTransactionSynchronization(){@OverridepublicvoidafterCommit(){// 事务提交成功后执行mqSender.send(message);}});}}

5. 运维重推接口

/** * 运维重推接口. * 用于手动触发失败事件的重新发布. */@RestController@RequestMapping("/api/inner/event")publicclassEventRetryController{@ResourceprivateEventMqSendereventMqSender;@ResourceprivateEventLogRepositoryeventLogRepository;@PostMapping("/retry")publicResult<Boolean>retry(@RequestBodyList<Integer>logIds){if(logIds==null||logIds.isEmpty()){// 未指定ID则查询所有失败记录List<EventLog>failedLogs=eventLogRepository.findByStatusInAndRetryCountLessThan(Arrays.asList(0,2),6);logIds=failedLogs.stream().map(EventLog::getId).collect(Collectors.toList());}logIds.forEach(eventMqSender::send);returnResult.success(true);}}

六、完整数据流示意

┌─ 你的服务内部 ─────────────────────────────┐ │ │ 业务代码 │ ① 写入本地消息表（status=0） │ │ │ ② 注册事务回调 │ ▼ │ │ │ 事务提交 ───────────│─────────┼────────────────────────────────────│ │ ▼ │ │ ③ 回调触发：发送 MQ（消息=logId） │ │ │ │ │ ▼ │ │ ④ MQ 消费者：读取消息表 → 组装 HTTP 请求 │ │ │ │ └─────────┼────────────────────────────────────┘ │ ▼ HTTP POST（CloudEvents 格式） ┌─────────────────────┐ │ 事件中心 │ │ 接收 → 路由 → 投递 │ └─────────┬───────────┘ │ Webhook 推送 ▼ ┌─────────────────────┐ │ b 服务 │ │ 接收事件 → 处理业务 │ └─────────────────────┘

七、核心概念速查表