事件驱动消息推送 - 快速开始

核心概念

使用Spring事件机制实现消息推送，业务系统只需发布事件，消息监听器自动保存消息到数据库。

快速部署

1. 无需额外部署

事件驱动架构已经集成到现有系统中，重启后端服务即可生效：

# Windows
bin\run.bat

# Linux
sh bin/run.sh

2. 验证部署

查看启动日志，确认以下内容：

... AsyncConfig           : Bean 'taskExecutor' created
... TaskMessageListener   : Bean created

使用方式

方式1: 在任务系统中（已集成）

任务创建、分配、状态变更时自动发布事件，无需额外代码

方式2: 在其他系统中使用

步骤1: 注入事件发布器

@Service
public class YourService {
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    // ...
}

步骤2: 发布事件

示例1: 任务创建时推送消息

public void createTask(Task task) {
    // 1. 业务逻辑
    taskMapper.insert(task);
    
    // 2. 发布事件（触发消息推送）
    eventPublisher.publishEvent(new TaskCreatedEvent(
        this,                    // source
        task.getTaskId(),        // 任务ID
        task.getTaskCode(),      // 任务编号
        task.getTaskType(),      // 任务类型
        task.getCreatorId(),     // 创建人ID
        "张三"                    // 创建人姓名
    ));
}

示例2: 任务分配时推送消息

public void assignTask(Long taskId, List<Long> assigneeIds) {
    // 1. 业务逻辑
    taskMapper.updateAssignees(taskId, assigneeIds);
    
    // 2. 发布事件（触发消息推送）
    eventPublisher.publishEvent(new TaskAssignedEvent(
        this,
        taskId,
        "TASK-001",
        assigneeIds,            // 执行人ID列表
        null,                   // 姓名列表（可选，监听器会查询）
        currentUserId,          // 分配人ID
        "李四"                  // 分配人姓名
    ));
}

示例3: 状态变更时推送消息

public void changeStatus(Long taskId, String newStatus) {
    Task task = taskMapper.selectById(taskId);
    String oldStatus = task.getStatus();
    
    // 1. 业务逻辑
    taskMapper.updateStatus(taskId, newStatus);
    
    // 2. 发布事件（触发消息推送）
    eventPublisher.publishEvent(new TaskStatusChangedEvent(
        this,
        taskId,
        task.getTaskCode(),
        oldStatus,              // 旧状态
        newStatus,              // 新状态
        "待处理",               // 旧状态描述
        "已完成",               // 新状态描述
        task.getAssigneeIds(),  // 执行人ID列表
        task.getCreatorId()     // 创建人ID
    ));
}

事件类型

事件类	用途	消息类型
TaskCreatedEvent	任务创建	CREATE
TaskAssignedEvent	任务分配	PUSH/ASSIGN
TaskStatusChangedEvent	状态变更	STATUS

扩展新事件

场景: 订单系统需要推送消息

1. 创建订单事件类

package com.ruoyi.system.event;

import org.springframework.context.ApplicationEvent;

public class OrderCreatedEvent extends ApplicationEvent {
    private Long orderId;
    private String orderNo;
    private Long customerId;
    
    public OrderCreatedEvent(Object source, Long orderId, String orderNo, Long customerId) {
        super(source);
        this.orderId = orderId;
        this.orderNo = orderNo;
        this.customerId = customerId;
    }
    
    // getters...
}

2. 创建监听器（或在现有监听器中添加）

@Component
public class OrderMessageListener {
    
    @Autowired
    private SysMessageMapper sysMessageMapper;
    
    @Async
    @EventListener
    public void handleOrderCreatedEvent(OrderCreatedEvent event) {
        SysMessage message = new SysMessage();
        message.setMessageType("ORDER_CREATE");
        message.setMessageTitle("订单创建成功");
        message.setMessageContent("您的订单" + event.getOrderNo() + "已创建");
        message.setReceiverId(event.getCustomerId());
        message.setIsRead("0");
        message.setCreateTime(DateUtils.getNowDate());
        message.setDelFlag("0");
        
        sysMessageMapper.insertSysMessage(message);
    }
}

3. 发布事件

@Service
public class OrderServiceImpl {
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    public void createOrder(Order order) {
        orderMapper.insert(order);
        
        // 发布事件
        eventPublisher.publishEvent(new OrderCreatedEvent(
            this, order.getId(), order.getOrderNo(), order.getCustomerId()
        ));
    }
}

核心优势

1. 完全解耦

// ❌ 旧方式：需要注入MessageService
@Autowired
private IMessageService messageService;

public void createTask() {
    // ...
    messageService.pushMessage(...);  // 强依赖
}

// ✅ 新方式：只需发布事件
@Autowired
private ApplicationEventPublisher eventPublisher;

public void createTask() {
    // ...
    eventPublisher.publishEvent(event);  // 零依赖
}

2. 异步处理

业务处理 200ms → 返回结果
    ↓（不阻塞）
发布事件 → 异步保存消息 50ms

总耗时：200ms（而非250ms）

3. 轻松扩展

添加新的消息推送场景，只需：
1. 发布事件 ✅
2. 无需修改业务代码 ✅
3. 无需重新部署 ✅

监控和调试

查看事件日志

# 搜索事件发布日志
grep "publishEvent" logs/ruoyi-admin.log

# 搜索事件监听日志
grep "TaskMessageListener" logs/ruoyi-admin.log

日志示例

2025-10-25 14:30:15 [main] INFO  发布任务创建事件，任务ID：1001
2025-10-25 14:30:15 [async-task-1] INFO  收到任务创建事件，任务ID：1001
2025-10-25 14:30:15 [async-task-1] INFO  任务创建消息已保存，消息ID：5001

常见问题

Q1: 事件发布后消息没有保存？

原因: 可能是异步线程池满了

解决:
1. 查看日志是否有异常
2. 调整线程池配置（AsyncConfig.java）
3. 检查数据库连接

Q2: 如何确保消息一定发送？

方案1: 使用同步监听器（去掉@Async）

@EventListener  // 不使用@Async
public void handleEvent(Event event) {
    // 同步执行，确保消息保存
}

方案2: 添加重试机制

@Async
@EventListener
@Retryable(maxAttempts = 3)
public void handleEvent(Event event) {
    // 失败自动重试3次
}

Q3: 如何禁用异步处理？

修改监听器，去掉 @Async 注解：

// @Async  // 注释掉
@EventListener
public void handleEvent(Event event) {
    // 现在是同步执行
}

性能优化

批量处理消息

// 收集消息，批量保存
List<SysMessage> messages = new ArrayList<>();
for (Long userId : userIds) {
    SysMessage msg = new SysMessage();
    // ...
    messages.add(msg);
}
sysMessageMapper.batchInsert(messages);  // 批量插入

调整线程池

根据实际负载调整 AsyncConfig.java:

// 高并发场景
executor.setCorePoolSize(10);
executor.setMaxPoolSize(20);

// 低并发场景
executor.setCorePoolSize(3);
executor.setMaxPoolSize(5);

对比方式

特性	直接调用	事件驱动
代码耦合	强依赖MessageService	零依赖
性能	同步阻塞	异步非阻塞
扩展性	需修改代码	只需添加监听器
测试	需模拟Service	只需发布事件

文件清单

新增文件

TaskEvent.java - 事件基类
TaskCreatedEvent.java - 创建事件
TaskAssignedEvent.java - 分配事件
TaskStatusChangedEvent.java - 状态事件
TaskMessageListener.java - 消息监听器
AsyncConfig.java - 异步配置

修改文件

SysTaskServiceImpl.java - 改为发布事件

下一步

📖 阅读详细文档：prd/事件驱动消息推送-实现总结.md
🔧 根据需要调整线程池配置
📊 监控系统运行日志
🚀 在其他系统中使用事件机制

更新时间: 2025-10-25
版本: v2.0
推荐使用: 适合所有需要消息推送的场景