事件驱动架构完整学习指南
作者: smj
日期: 2025-01-20
版本: V1.0
项目: HCMS-WMS 机器人管理模块 V3.0
📚 目录
-
核心概念
-
三大核心类详解
-
完整工作流程
-
代码实战示例
-
设计模式深入
-
性能优化要点
-
常见问题FAQ
-
扩展知识
🎯 核心理解图
监听器实现接口 → 重写propertyChange()方法 → 注册到事件总线 ↓ 事件总线存储监听器 ↓业务类发布事件 → 调用事件总线的firePropertyChange() ↓ 事件总线遍历所有注册的监听器 ↓ 匹配事件模式 ↓ 调用监听器的propertyChange()方法 ↓ 执行业务逻辑核心概念
什么是事件驱动架构？
事件驱动架构（Event-Driven Architecture, EDA）是一种软件设计模式，系统中的组件通过发布事件和订阅事件来进行通信，而不是直接调用彼此的方法。
核心角色
┌─────────────┐ 发布事件 ┌─────────────┐ 通知 ┌─────────────┐
│ 发布者 │ ───────────────→ │ 事件总线 │ ───────────→ │ 监听器 │
│ (Publisher) │ │ (Event Bus) │ │ (Listener) │
└─────────────┘ └─────────────┘ └─────────────┘
业务服务类 中介/容器 处理业务逻辑
三大角色：
-
发布者（Publisher）：发布事件的组件（如业务服务类）
-
事件总线（Event Bus）：管理监听器和分发事件的中介
-
监听器（Listener）：订阅事件并处理业务逻辑的组件
为什么使用事件驱动架构？
优点：
-
✅ 解耦：发布者和监听器互不依赖，降低耦合度
-
✅ 扩展性：新增监听器不影响现有代码
-
✅ 异步处理：事件发布后立即返回，不阻塞主流程
-
✅ 灵活性：一个事件可以被多个监听器处理
缺点：
-
❌ 调试困难：事件流程不直观，需要追踪事件链路
-
❌ 性能开销：事件分发和异步处理有一定开销
-
❌ 事件丢失风险：需要持久化机制保证可靠性
三大核心类详解
1. WmsEventBus - 事件总线
职责
-
管理所有监听器（注册/移除）
-
接收事件发布请求
-
匹配事件模式
-
异步通知监听器
核心数据结构
// 存储监听器的容器
private final Map<String, List> listeners = new ConcurrentHashMap<>();
// ↑ ↑
// 事件模式（Key） 监听器列表（Value）
// 示例数据：
// {
// "wms:in:*": [RobotScheduleListener, LoggingListener],
// "wms:out:*": [RobotScheduleListener, InventoryListener],
// "wms:robot:*": [MonitoringListener]
// }
为什么用 ConcurrentHashMap？
-
多线程环境下保证线程安全
-
支持高并发读写操作
-
避免数据竞争和不一致
核心方法
1. 注册监听器
public void addPropertyChangeListener(String pattern, PropertyChangeListener listener) {
listeners.computeIfAbsent(pattern, k -> new CopyOnWriteArrayList<>()).add(listener);
log.info("监听器已注册: pattern={}, listener={}", pattern, listener.getClass().getSimpleName());
}
关键点：
-
computeIfAbsent：如果key不存在，创建新列表；存在则返回现有列表 -
CopyOnWriteArrayList：写时复制列表，适合读多写少场景 -
支持同一个监听器订阅多个事件模式
2. 发布事件
public void firePropertyChange(String eventName, Object oldValue, Object newValue) {
PropertyChangeEvent event = new PropertyChangeEvent(this, eventName, oldValue, newValue);
// 遍历所有注册的监听器
for (Map.Entry<String, List> entry : listeners.entrySet()) {
String pattern = entry.getKey();
// 匹配事件模式
if (matches(pattern, eventName)) {
// 异步通知每个监听器
for (PropertyChangeListener listener : entry.getValue()) {
executor.submit(() -> notifyListener(listener, event));
}
}
}
}
执行流程：
1. 创建事件对象 ↓2. 遍历所有监听器 ↓3. 匹配事件模式（通配符支持） ↓4. 使用虚拟线程异步通知 ↓5. 调用监听器的 propertyChange() 方法3. 通配符匹配
private boolean matches(String pattern, String eventName) {
if (pattern.equals(eventName)) {
return true; // 完全匹配
}
if (pattern.endsWith("*")) {
String prefix = pattern.substring(0, pattern.length() - 1);
return eventName.startsWith(prefix); // 前缀匹配
}
return false;
}
匹配规则：
matches("wms:in:*", "wms:in:start") → true ✅ 前缀匹配matches("wms:in:*", "wms:in:complete") → true ✅ 前缀匹配matches("wms:in:*", "wms:out:start") → false ❌ 前缀不匹配matches("wms:in:start", "wms:in:start") → true ✅ 完全匹配虚拟线程优化
// 构造方法中初始化线程池
public WmsEventBus() {
ExecutorService executor;
try {
// Java 21+ 使用虚拟线程
Method virtualThreadMethod = Executors.class.getMethod("newVirtualThreadPerTaskExecutor");
executor = (ExecutorService) virtualThreadMethod.invoke(null);
log.info("使用虚拟线程池");
} catch (Exception e) {
// 降级到普通线程池
executor = Executors.newFixedThreadPool(10);
log.info("使用普通线程池");
}
this.executor = executor;
}
虚拟线程 vs 普通线程：
| 特性 | 虚拟线程 | 普通线程 |
|------|---------|---------|
| 创建成本 | 极低 | 高 |
| 内存占用 | 几KB | 1-2MB |
| 数量限制 | 百万级 | 几千个 |
| 适用场景 | 高并发I/O | CPU密集型 |
2. WmsEventPersistence - 事件持久化服务
职责
-
将事件保存到数据库
-
提供故障恢复能力
-
支持事件重试机制
-
记录事件处理状态
核心方法
1. 保存事件
@Transactional(rollbackFor = Exception.class)
public String saveEvent(String eventName, String businessId, String businessType, String eventData) {
WmsEventLog eventLog = new WmsEventLog();
eventLog.setEventName(eventName);
eventLog.setBusinessId(businessId); // 业务单号
eventLog.setBusinessType(businessType); // 业务类型
eventLog.setEventData(eventData); // 事件数据（JSON）
eventLog.setStatus(EventStatusEnum.PENDING.getCode());
eventLog.setRetryCount(0);
eventLogMapper.insert(eventLog);
return eventLog.getId();
}
使用场景：
// 入库单开始时保存事件String eventId = wmsEventPersistence.saveEvent("wms:in:start", // 事件名称"IN202501200001", // 入库单号"INBOUND", // 业务类型JSON.toJSONString(order) // 事件数据);2. 更新事件状态
@Transactional(rollbackFor = Exception.class)
public void updateEventStatus(String eventId, EventStatusEnum status, String errorMessage) {
WmsEventLog eventLog = eventLogMapper.selectById(eventId);
eventLog.setStatus(status.getCode());
eventLog.setUpdateTime(new Date());
if (status == EventStatusEnum.FAILED && errorMessage != null) {
eventLog.setErrorMessage(errorMessage);
}
if (status == EventStatusEnum.COMPLETED) {
eventLog.setProcessedTime(new Date());
}
eventLogMapper.updateById(eventLog);
}
状态流转：
PENDING（待处理） ↓PROCESSING（处理中） ↓COMPLETED（已完成） / FAILED（失败）3. 扫描未处理事件（故障恢复）
public List scanUnprocessedEvents(int maxRetryCount, int limit) {
LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(WmsEventLog::getStatus, EventStatusEnum.PENDING.getCode())
.lt(WmsEventLog::getRetryCount, maxRetryCount)
.orderByAsc(WmsEventLog::getCreateTime)
.last("LIMIT " + limit);
return eventLogMapper.selectList(queryWrapper);
}
使用场景：
// 系统重启后，扫描未处理的事件重新处理@Scheduled(fixedDelay = 60000) // 每分钟执行一次public void recoverUnprocessedEvents() {List<WmsEventLog> events = wmsEventPersistence.scanUnprocessedEvents(3, 100);for (WmsEventLog event : events) {// 重新发布事件wmsEventBus.firePropertyChange(event.getEventName(), null, event.getEventData());// 更新状态为处理中wmsEventPersistence.updateEventStatus(event.getId(), EventStatusEnum.PROCESSING, null);}}4. 重试失败事件
@Transactional(rollbackFor = Exception.class)
public boolean retryEvent(String eventId) {
WmsEventLog eventLog = eventLogMapper.selectById(eventId);
// 增加重试次数
eventLog.setRetryCount(eventLog.getRetryCount() + 1);
// 重置状态为PENDING
eventLog.setStatus(EventStatusEnum.PENDING.getCode());
eventLog.setUpdateTime(new Date());
eventLogMapper.updateById(eventLog);
return true;
}
---### 3. WmsEventNames - 事件名称常量#### 职责- 统一管理所有事件名称- 避免硬编码和拼写错误- 提供工具方法判断事件类型#### 命名规范wms:{业务模块}:{操作}
**示例**：- `wms:in:start` - 入库开始- `wms:out:complete` - 出库完成- `wms:robot:task:finish` - 机器人任务完成#### 核心常量// 入库相关public static final String INBOUND_START = "wms:in:start";public static final String INBOUND_COMPLETE = "wms:in:complete";public static final String INBOUND_CANCEL = "wms:in:cancel";// 出库相关public static final String OUTBOUND_START = "wms:out:start";public static final String OUTBOUND_COMPLETE = "wms:out:complete";// 机器人相关public static final String ROBOT_TASK_FINISH = "wms:robot:task:finish"; // 重要！触发自动调度// 通配符模式public static final String INBOUND_ALL = "wms:in:*";public static final String ROBOT_ALL = "wms:robot:*";工具方法
// 判断是否为业务事件public static boolean isBusinessEvent(String eventName) {return eventName.startsWith("wms:in:")|| eventName.startsWith("wms:out:")|| eventName.startsWith("wms:check:");}// 提取业务类型public static String extractBusinessType(String eventName) {if (eventName.startsWith("wms:in:")) return "INBOUND";if (eventName.startsWith("wms:out:")) return "OUTBOUND";return "UNKNOWN";}完整工作流程
流程图
[系统启动]
↓
[监听器注册]
RobotScheduleListener.init()
wmsEventBus.addPropertyChangeListener("wms:in:*", this)
↓
[事件总线存储]
listeners = {
"wms:in:*": [RobotScheduleListener]
}
↓
[用户操作]
POST /api/inbound/create
↓
[业务处理]
InboundService.createInbound()
保存入库单到数据库
↓
[发布事件]
wmsEventBus.firePropertyChange("wms:in:start", null, order)
↓
[事件总线匹配]
遍历listeners
匹配"wms:in:*"和"wms:in:start" ✅
↓
[异步通知]
executor.submit(() -> {
RobotScheduleListener.propertyChange(event)
})
↓
[监听器处理]
handleInboundEvent()
查询可用机器人
创建机器人任务
下发任务
### 时序图用户 Controller Service EventBus Listener
│ │ │ │ │
│─创建入库单──→│ │ │ │
│ │─调用服务──→│ │ │
│ │ │─保存数据 │ │
│ │ │─发布事件───→│ │
│ │ │ │─匹配模式 │
│ │ │ │─通知监听器→│
│ │ │ │ │─处理业务
│ │←返回成功────│ │ │
│←返回结果────│ │ │ │
---## 代码实战示例### 完整示例：入库流程#### 1. 监听器类package com.slhc.hcms.module.robot.listener;import com.slhc.hcms.module.robot.event.WmsEventBus;import com.slhc.hcms.module.robot.service.IRobotInfoService;import com.slhc.hcms.module.robot.service.IRobotTaskService;import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;import java.beans.PropertyChangeEvent;import java.beans.PropertyChangeListener;@Slf4j@Componentpublic class RobotScheduleListener implements PropertyChangeListener { @Autowired private WmsEventBus wmsEventBus; @Autowired private IRobotInfoService robotInfoService; @Autowired private IRobotTaskService robotTaskService; // 系统启动时自动注册 @PostConstruct public void init() { log.info("========== 开始注册机器人调度监听器 =========="); // 订阅入库事件 wmsEventBus.addPropertyChangeListener("wms:in:*", this); log.info("✅ 已订阅入库事件"); // 订阅出库事件 wmsEventBus.addPropertyChangeListener("wms:out:*", this); log.info("✅ 已订阅出库事件"); // 订阅盘点事件 wmsEventBus.addPropertyChangeListener("wms:check:*", this); log.info("✅ 已订阅盘点事件"); log.info("========== 监听器注册完成 =========="); } // 当事件发生时，这个方法会被自动调用 @Override public void propertyChange(PropertyChangeEvent evt) { String eventName = evt.getPropertyName(); Object eventData = evt.getNewValue(); log.info("🔔 收到事件: {}", eventName); try { // 根据事件类型分发处理 if (eventName.startsWith("wms:in:")) { handleInboundEvent(eventName, eventData); } else if (eventName.startsWith("wms:out:")) { handleOutboundEvent(eventName, eventData); } else if (eventName.startsWith("wms:check:")) { handleCheckEvent(eventName, eventData); } } catch (Exception e) { log.error("处理事件失败: eventName={}, error={}", eventName, e.getMessage(), e); } } // 处理入库事件 private void handleInboundEvent(String eventName, Object eventData) { log.info("📥 处理入库事件: {}", eventName); // 1. 查询可用机器人 RobotInfo robot = robotInfoService.findAvailableRobot("WH01"); if (robot == null) { log.warn("⚠️ 没有可用机器人"); return; } // 2. 创建机器人任务 RobotTask task = robotTaskService.createTask( robot.getId(), "INBOUND", eventData.toString() ); log.info("✅ 机器人任务已创建: taskId={}, robotCode={}", task.getId(), robot.getRobotCode()); } // 处理出库事件 private void handleOutboundEvent(String eventName, Object eventData) { log.info("📤 处理出库事件: {}", eventName); // 类似入库处理逻辑 } // 处理盘点事件 private void handleCheckEvent(String eventName, Object eventData) { log.info("📋 处理盘点事件: {}", eventName); // 类似入库处理逻辑 }}2. 业务服务类
package com.slhc.hcms.module.inbound.service.impl;import com.slhc.hcms.module.inbound.entity.InboundOrder;import com.slhc.hcms.module.robot.event.WmsEventBus;import com.slhc.hcms.module.robot.event.WmsEventNames;import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;@Slf4j@Servicepublic class InboundServiceImpl {@Autowiredprivate WmsEventBus wmsEventBus;@Transactional(rollbackFor = Exception.class)public void createInbound(InboundOrder order) {log.info("========== 开始创建入库单 ==========");log.info("入库单号: {}", order.getOrderNo());// 1. 保存入库单到数据库saveToDatabase(order);log.info("💾 入库单已保存");// 2. 发布事件（通知其他模块）wmsEventBus.firePropertyChange(WmsEventNames.INBOUND_START, // 使用常量null,order);log.info("📢 入库开始事件已发布");log.info("========== 入库单创建完成 ==========");}private void saveToDatabase(InboundOrder order) {// 保存逻辑}}3. 控制器类
package com.slhc.hcms.module.inbound.controller;import com.slhc.hcms.module.inbound.entity.InboundOrder;import com.slhc.hcms.module.inbound.service.impl.InboundServiceImpl;import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.*;@Slf4j@RestController@RequestMapping("/api/inbound")public class InboundController {@Autowiredprivate InboundServiceImpl inboundService;@PostMapping("/create")public Result createInbound(@RequestBody InboundOrder order) {log.info("收到创建入库单请求: {}", order.getOrderNo());inboundService.createInbound(order);return Result.success("入库单创建成功");}}设计模式深入
观察者模式（Observer Pattern）
事件驱动架构是观察者模式的一种实现。
角色映射：
-
Subject（主题） = Event Bus（事件总线）
-
Observer（观察者） = Listener（监听器）
-
notify() = firePropertyChange()
类图：
┌─────────────────┐│ EventBus │├─────────────────┤│ +addListener() ││ +fireEvent() │└────────┬────────┘ │ 1 │ │ *┌────────▼────────┐│ Listener │├─────────────────┤│ +propertyChange()│└─────────────────┘发布-订阅模式（Pub-Sub Pattern）
与观察者模式的区别：
| 特性 | 观察者模式 | 发布-订阅模式 |
|------|-----------|--------------|
| 耦合度 | 主题知道观察者 | 发布者不知道订阅者 |
| 中介 | 无 | 有（事件总线） |
| 通信方式 | 直接调用 | 通过中介 |
| 灵活性 | 较低 | 较高 |
💡 我们的实现更接近发布-订阅模式，因为有事件总线作为中介。
性能优化要点
1. 使用虚拟线程
// Java 21+ 虚拟线程ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();// 优点：// - 创建成本极低// - 可以创建百万级线程// - 适合I/O密集型任务2. 异步处理
// 事件发布后立即返回，不等待监听器处理完成wmsEventBus.firePropertyChange("wms:in:start", null, order);// 这里立即返回，监听器在后台异步处理3. 异常隔离
// 一个监听器异常不影响其他监听器for (PropertyChangeListener listener : listeners) {try {listener.propertyChange(event);} catch (Exception e) {log.error("监听器处理失败: {}", e.getMessage());// 继续处理下一个监听器}}4. 事件持久化
// 保证事件不丢失wmsEventPersistence.saveEvent(eventName, businessId, businessType, eventData);// 定时扫描未处理事件@Scheduled(fixedDelay = 60000)public void recoverEvents() {List<WmsEventLog> events = wmsEventPersistence.scanUnprocessedEvents(3, 100);// 重新处理}常见问题FAQ
Q1: PropertyChangeListener是什么？
A: 这是Java标准库提供的一个接口（不是方法），定义了事件监听的契约。
package java.beans;public interface PropertyChangeListener {void propertyChange(PropertyChangeEvent evt);}任何实现这个接口的类都必须提供propertyChange()方法。
Q2: 为什么要重写propertyChange()方法？
A: 因为事件总线需要调用这个方法来通知监听器。
// 事件总线内部listener.propertyChange(event); // 调用你重写的方法Q3: 监听器和事件是什么关系？
A:
-
❌ 不是"监听器包含事件"
-
✅ 是"监听器订阅事件"
类比：
-
事件 = 电视节目
-
监听器 = 观众
-
订阅 = 观众选择看哪些节目
Q4: 一个监听器可以监听多个事件吗？
A: ✅ 可以！
@PostConstructpublic void init() {wmsEventBus.addPropertyChangeListener("wms:in:*", this);wmsEventBus.addPropertyChangeListener("wms:out:*", this);wmsEventBus.addPropertyChangeListener("wms:check:*", this);}Q5: 事件总线怎么通知监听器？
A: 通过接口调用
// 事件总线持有监听器对象的引用PropertyChangeListener listener = robotScheduleListener;// 通过接口调用监听器的方法listener.propertyChange(event); // ← 这就是通知！Q6: 注册监听器和虚拟线程有什么关系？
A:
-
注册监听器时：不使用线程（只是把监听器放到Map里）
-
发布事件时：使用虚拟线程异步通知监听器
Q7: 事件会丢失吗？
A: 可能会。解决方案：
-
事件持久化：保存到数据库
-
定时扫描：扫描未处理事件重新处理
-
重试机制：失败事件自动重试
扩展知识
1. Spring的事件机制
Spring也提供了事件驱动机制，与我们的实现类似：
// Spring事件
public class InboundStartEvent extends ApplicationEvent {
public InboundStartEvent(Object source) {
super(source);
}
}
// Spring监听器
@Component
public class RobotScheduleListener {
@EventListener
public void handleInboundStart(InboundStartEvent event) {
// 处理业务
}
}
// 发布事件
@Autowired
private ApplicationEventPublisher eventPublisher;
eventPublisher.publishEvent(new InboundStartEvent(order));
**对比**：| 特性 | 我们的实现 | Spring事件 ||------|-----------|-----------|| 通配符支持 | ✅ | ❌ || 异步处理 | ✅ | 需要@Async || 事件持久化 | ✅ | ❌ || 学习成本 | 低 | 中 |---### 2. 消息队列（MQ）事件驱动架构的进阶版本是使用消息队列：发布者 → RabbitMQ/Kafka → 订阅者
```
优点：
-
✅ 更高的可靠性
-
✅ 支持分布式
-
✅ 消息持久化
-
✅ 流量削峰
缺点：
-
❌ 部署复杂
-
❌ 运维成本高
总结
通过本文档，你应该已经掌握了：
-
✅ 事件驱动架构的核心概念和工作原理
-
✅ 三大核心类（WmsEventBus、WmsEventPersistence、WmsEventNames）的详细实现
-
✅ 完整的事件发布-订阅流程
-
✅ 实战代码示例和最佳实践
-
✅ 性能优化技巧和常见问题解决方案
-
✅ 与Spring事件机制和消息队列的对比
关键要点回顾：
-
PropertyChangeListener 是Java标准库的接口，不是方法
-
监听器 有两个职责：注册自己（@PostConstruct）+ 处理业务（propertyChange方法）
-
事件总线 通过接口调用通知监听器
-
虚拟线程 用于异步通知，注册时不使用线程
-
通配符匹配 支持灵活的事件订阅模式
