👋 大家好，我是 阿问学长！专注于分享优质开源项目解析、毕业设计项目指导支持、幼小初高的教辅资料推荐等，欢迎关注交流！🚀

Spring事件机制

前言

Spring事件机制是基于观察者模式的实现，提供了一种松耦合的方式来处理应用程序中的事件。通过事件发布与监听机制，可以实现模块间的解耦，提高系统的可维护性和扩展性。RuoYi-Vue-Plus框架充分利用了Spring事件机制来处理各种业务场景。本文将详细介绍事件发布与监听、异步事件处理、自定义事件实现以及事件驱动架构等内容。

事件发布与监听

基础事件定义

/**
 * 用户事件基类
 */
public abstract class UserEvent extends ApplicationEvent {
    
    private final Long userId;
    private final String username;
    private final Date timestamp;
    
    public UserEvent(Object source, Long userId, String username) {
        super(source);
        this.userId = userId;
        this.username = username;
        this.timestamp = new Date();
    }
    
    public Long getUserId() {
        return userId;
    }
    
    public String getUsername() {
        return username;
    }
    
    public Date getTimestamp() {
        return timestamp;
    }
}

/**
 * 用户登录事件
 */
public class UserLoginEvent extends UserEvent {
    
    private final String ipAddress;
    private final String userAgent;
    private final String loginType;
    
    public UserLoginEvent(Object source, Long userId, String username, 
                         String ipAddress, String userAgent, String loginType) {
        super(source, userId, username);
        this.ipAddress = ipAddress;
        this.userAgent = userAgent;
        this.loginType = loginType;
    }
    
    public String getIpAddress() {
        return ipAddress;
    }
    
    public String getUserAgent() {
        return userAgent;
    }
    
    public String getLoginType() {
        return loginType;
    }
}

/**
 * 用户注销事件
 */
public class UserLogoutEvent extends UserEvent {
    
    private final String reason;
    
    public UserLogoutEvent(Object source, Long userId, String username, String reason) {
        super(source, userId, username);
        this.reason = reason;
    }
    
    public String getReason() {
        return reason;
    }
}

/**
 * 用户注册事件
 */
public class UserRegistrationEvent extends UserEvent {
    
    private final String email;
    private final String phone;
    
    public UserRegistrationEvent(Object source, Long userId, String username, 
                                String email, String phone) {
        super(source, userId, username);
        this.email = email;
        this.phone = phone;
    }
    
    public String getEmail() {
        return email;
    }
    
    public String getPhone() {
        return phone;
    }
}

/**
 * 数据变更事件
 */
public class DataChangeEvent extends ApplicationEvent {
    
    private final String tableName;
    private final String operation;
    private final Object oldData;
    private final Object newData;
    private final Long operatorId;
    
    public DataChangeEvent(Object source, String tableName, String operation, 
                          Object oldData, Object newData, Long operatorId) {
        super(source);
        this.tableName = tableName;
        this.operation = operation;
        this.oldData = oldData;
        this.newData = newData;
        this.operatorId = operatorId;
    }
    
    public String getTableName() {
        return tableName;
    }
    
    public String getOperation() {
        return operation;
    }
    
    public Object getOldData() {
        return oldData;
    }
    
    public Object getNewData() {
        return newData;
    }
    
    public Long getOperatorId() {
        return operatorId;
    }
}

事件监听器

/**
 * 用户事件监听器
 */
@Component
@Slf4j
public class UserEventListener {
    
    @Autowired
    private ISysLogininforService logininforService;
    
    @Autowired
    private NotificationService notificationService;
    
    @Autowired
    private SecurityService securityService;
    
    /**
     * 监听用户登录事件
     */
    @EventListener
    public void handleUserLogin(UserLoginEvent event) {
        log.info("用户登录事件: 用户ID={}, 用户名={}, IP={}", 
                event.getUserId(), event.getUsername(), event.getIpAddress());
        
        try {
            // 记录登录日志
            SysLogininfor logininfor = new SysLogininfor();
            logininfor.setUserName(event.getUsername());
            logininfor.setIpaddr(event.getIpAddress());
            logininfor.setBrowser(UserAgent.parseUserAgentString(event.getUserAgent()).getBrowser().getName());
            logininfor.setOs(UserAgent.parseUserAgentString(event.getUserAgent()).getOperatingSystem().getName());
            logininfor.setStatus(Constants.LOGIN_SUCCESS);
            logininfor.setMsg("登录成功");
            logininfor.setLoginTime(event.getTimestamp());
            
            logininforService.insertLogininfor(logininfor);
            
            // 安全检查
            securityService.checkLoginSecurity(event);
            
        } catch (Exception e) {
            log.error("处理用户登录事件失败", e);
        }
    }
    
    /**
     * 监听用户注销事件
     */
    @EventListener
    public void handleUserLogout(UserLogoutEvent event) {
        log.info("用户注销事件: 用户ID={}, 用户名={}, 原因={}", 
                event.getUserId(), event.getUsername(), event.getReason());
        
        try {
            // 记录注销日志
            SysLogininfor logininfor = new SysLogininfor();
            logininfor.setUserName(event.getUsername());
            logininfor.setStatus(Constants.LOGOUT);
            logininfor.setMsg("注销成功: " + event.getReason());
            logininfor.setLoginTime(event.getTimestamp());
            
            logininforService.insertLogininfor(logininfor);
            
            // 清理用户会话相关数据
            securityService.cleanupUserSession(event.getUserId());
            
        } catch (Exception e) {
            log.error("处理用户注销事件失败", e);
        }
    }
    
    /**
     * 监听用户注册事件
     */
    @EventListener
    public void handleUserRegistration(UserRegistrationEvent event) {
        log.info("用户注册事件: 用户ID={}, 用户名={}, 邮箱={}", 
                event.getUserId(), event.getUsername(), event.getEmail());
        
        try {
            // 发送欢迎邮件
            notificationService.sendWelcomeEmail(event.getEmail(), event.getUsername());
            
            // 发送欢迎短信
            if (StringUtils.hasText(event.getPhone())) {
                notificationService.sendWelcomeSms(event.getPhone(), event.getUsername());
            }
            
            // 初始化用户默认设置
            initializeUserDefaults(event.getUserId());
            
        } catch (Exception e) {
            log.error("处理用户注册事件失败", e);
        }
    }
    
    /**
     * 条件监听 - 只监听特定类型的登录
     */
    @EventListener(condition = "#event.loginType == 'WEB'")
    public void handleWebLogin(UserLoginEvent event) {
        log.info("Web登录事件: 用户={}", event.getUsername());
        
        // Web登录特殊处理逻辑
        handleWebSpecificLogin(event);
    }
    
    /**
     * 条件监听 - 只监听移动端登录
     */
    @EventListener(condition = "#event.loginType == 'MOBILE'")
    public void handleMobileLogin(UserLoginEvent event) {
        log.info("移动端登录事件: 用户={}", event.getUsername());
        
        // 移动端登录特殊处理逻辑
        handleMobileSpecificLogin(event);
    }
    
    /**
     * 初始化用户默认设置
     */
    private void initializeUserDefaults(Long userId) {
        // 实现用户默认设置初始化逻辑
        log.debug("初始化用户默认设置: userId={}", userId);
    }
    
    /**
     * Web登录特殊处理
     */
    private void handleWebSpecificLogin(UserLoginEvent event) {
        // 实现Web登录特殊处理逻辑
        log.debug("处理Web登录特殊逻辑");
    }
    
    /**
     * 移动端登录特殊处理
     */
    private void handleMobileSpecificLogin(UserLoginEvent event) {
        // 实现移动端登录特殊处理逻辑
        log.debug("处理移动端登录特殊逻辑");
    }
}

/**
 * 数据变更事件监听器
 */
@Component
@Slf4j
public class DataChangeEventListener {
    
    @Autowired
    private CacheManager cacheManager;
    
    @Autowired
    private SearchIndexService searchIndexService;
    
    @Autowired
    private AuditService auditService;
    
    /**
     * 监听数据变更事件
     */
    @EventListener
    public void handleDataChange(DataChangeEvent event) {
        log.info("数据变更事件: 表={}, 操作={}, 操作人={}", 
                event.getTableName(), event.getOperation(), event.getOperatorId());
        
        try {
            // 记录审计日志
            auditService.recordDataChange(event);
            
            // 更新缓存
            updateCache(event);
            
            // 更新搜索索引
            updateSearchIndex(event);
            
        } catch (Exception e) {
            log.error("处理数据变更事件失败", e);
        }
    }
    
    /**
     * 监听用户表变更
     */
    @EventListener(condition = "#event.tableName == 'sys_user'")
    public void handleUserDataChange(DataChangeEvent event) {
        log.info("用户数据变更: 操作={}", event.getOperation());
        
        try {
            // 清理用户相关缓存
            clearUserCache(event);
            
            // 同步用户信息到其他系统
            syncUserToExternalSystems(event);
            
        } catch (Exception e) {
            log.error("处理用户数据变更失败", e);
        }
    }
    
    /**
     * 更新缓存
     */
    private void updateCache(DataChangeEvent event) {
        String cacheName = "data:" + event.getTableName();
        Cache cache = cacheManager.getCache(cacheName);
        
        if (cache != null) {
            if ("DELETE".equals(event.getOperation())) {
                cache.clear();
            } else {
                // 清理相关缓存项
                cache.evict(event.getNewData());
            }
        }
    }
    
    /**
     * 更新搜索索引
     */
    private void updateSearchIndex(DataChangeEvent event) {
        if ("INSERT".equals(event.getOperation()) || "UPDATE".equals(event.getOperation())) {
            searchIndexService.indexDocument(event.getTableName(), event.getNewData());
        } else if ("DELETE".equals(event.getOperation())) {
            searchIndexService.deleteDocument(event.getTableName(), event.getOldData());
        }
    }
    
    /**
     * 清理用户缓存
     */
    private void clearUserCache(DataChangeEvent event) {
        // 实现用户缓存清理逻辑
        log.debug("清理用户缓存");
    }
    
    /**
     * 同步用户信息到外部系统
     */
    private void syncUserToExternalSystems(DataChangeEvent event) {
        // 实现用户信息同步逻辑
        log.debug("同步用户信息到外部系统");
    }
}

事件发布器

/**
 * 事件发布服务
 */
@Service
@Slf4j
public class EventPublisherService {
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    /**
     * 发布用户登录事件
     */
    public void publishUserLoginEvent(Long userId, String username, 
                                    String ipAddress, String userAgent, String loginType) {
        try {
            UserLoginEvent event = new UserLoginEvent(this, userId, username, 
                                                     ipAddress, userAgent, loginType);
            eventPublisher.publishEvent(event);
            
            log.debug("发布用户登录事件: userId={}, username={}", userId, username);
            
        } catch (Exception e) {
            log.error("发布用户登录事件失败", e);
        }
    }
    
    /**
     * 发布用户注销事件
     */
    public void publishUserLogoutEvent(Long userId, String username, String reason) {
        try {
            UserLogoutEvent event = new UserLogoutEvent(this, userId, username, reason);
            eventPublisher.publishEvent(event);
            
            log.debug("发布用户注销事件: userId={}, username={}", userId, username);
            
        } catch (Exception e) {
            log.error("发布用户注销事件失败", e);
        }
    }
    
    /**
     * 发布用户注册事件
     */
    public void publishUserRegistrationEvent(Long userId, String username, 
                                           String email, String phone) {
        try {
            UserRegistrationEvent event = new UserRegistrationEvent(this, userId, username, 
                                                                   email, phone);
            eventPublisher.publishEvent(event);
            
            log.debug("发布用户注册事件: userId={}, username={}", userId, username);
            
        } catch (Exception e) {
            log.error("发布用户注册事件失败", e);
        }
    }
    
    /**
     * 发布数据变更事件
     */
    public void publishDataChangeEvent(String tableName, String operation, 
                                     Object oldData, Object newData, Long operatorId) {
        try {
            DataChangeEvent event = new DataChangeEvent(this, tableName, operation, 
                                                       oldData, newData, operatorId);
            eventPublisher.publishEvent(event);
            
            log.debug("发布数据变更事件: table={}, operation={}", tableName, operation);
            
        } catch (Exception e) {
            log.error("发布数据变更事件失败", e);
        }
    }
}

异步事件处理

异步事件配置

/**
 * 异步事件配置
 */
@Configuration
@EnableAsync
public class AsyncEventConfig {
    
    /**
     * 异步事件执行器
     */
    @Bean("asyncEventExecutor")
    public TaskExecutor asyncEventExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("async-event-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
    
    /**
     * 异步事件监听器异常处理器
     */
    @Bean
    public AsyncUncaughtExceptionHandler asyncUncaughtExceptionHandler() {
        return new SimpleAsyncUncaughtExceptionHandler();
    }
    
    /**
     * 配置异步执行器
     */
    @Override
    public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
        configurer.setTaskExecutor(asyncEventExecutor());
        configurer.setDefaultTimeout(30000);
    }
}

/**
 * 异步事件监听器
 */
@Component
@Slf4j
public class AsyncEventListener {
    
    @Autowired
    private EmailService emailService;
    
    @Autowired
    private SmsService smsService;
    
    @Autowired
    private StatisticsService statisticsService;
    
    /**
     * 异步处理用户登录事件
     */
    @Async("asyncEventExecutor")
    @EventListener
    public void handleUserLoginAsync(UserLoginEvent event) {
        log.info("异步处理用户登录事件: 用户={}", event.getUsername());
        
        try {
            // 模拟耗时操作
            Thread.sleep(1000);
            
            // 更新用户登录统计
            statisticsService.updateLoginStatistics(event.getUserId(), event.getTimestamp());
            
            // 检查异常登录
            checkAbnormalLogin(event);
            
            log.info("异步处理用户登录事件完成: 用户={}", event.getUsername());
            
        } catch (Exception e) {
            log.error("异步处理用户登录事件失败", e);
        }
    }
    
    /**
     * 异步发送注册欢迎邮件
     */
    @Async("asyncEventExecutor")
    @EventListener
    public void sendWelcomeEmailAsync(UserRegistrationEvent event) {
        log.info("异步发送欢迎邮件: 用户={}, 邮箱={}", event.getUsername(), event.getEmail());
        
        try {
            // 模拟邮件发送耗时
            Thread.sleep(2000);
            
            emailService.sendWelcomeEmail(event.getEmail(), event.getUsername());
            
            log.info("异步发送欢迎邮件完成: 用户={}", event.getUsername());
            
        } catch (Exception e) {
            log.error("异步发送欢迎邮件失败", e);
        }
    }
    
    /**
     * 异步发送注册欢迎短信
     */
    @Async("asyncEventExecutor")
    @EventListener
    public void sendWelcomeSmsAsync(UserRegistrationEvent event) {
        if (!StringUtils.hasText(event.getPhone())) {
            return;
        }
        
        log.info("异步发送欢迎短信: 用户={}, 手机={}", event.getUsername(), event.getPhone());
        
        try {
            // 模拟短信发送耗时
            Thread.sleep(1500);
            
            smsService.sendWelcomeSms(event.getPhone(), event.getUsername());
            
            log.info("异步发送欢迎短信完成: 用户={}", event.getUsername());
            
        } catch (Exception e) {
            log.error("异步发送欢迎短信失败", e);
        }
    }
    
    /**
     * 异步处理数据变更统计
     */
    @Async("asyncEventExecutor")
    @EventListener
    public void handleDataChangeStatisticsAsync(DataChangeEvent event) {
        log.info("异步处理数据变更统计: 表={}, 操作={}", event.getTableName(), event.getOperation());
        
        try {
            // 更新数据变更统计
            statisticsService.updateDataChangeStatistics(
                event.getTableName(), 
                event.getOperation(), 
                event.getOperatorId()
            );
            
            log.info("异步处理数据变更统计完成: 表={}", event.getTableName());
            
        } catch (Exception e) {
            log.error("异步处理数据变更统计失败", e);
        }
    }
    
    /**
     * 检查异常登录
     */
    private void checkAbnormalLogin(UserLoginEvent event) {
        // 实现异常登录检查逻辑
        log.debug("检查异常登录: 用户={}, IP={}", event.getUsername(), event.getIpAddress());
    }
}

/**
 * 自定义异步异常处理器
 */
@Component
@Slf4j
public class CustomAsyncUncaughtExceptionHandler implements AsyncUncaughtExceptionHandler {
    
    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        log.error("异步事件处理异常: 方法={}, 参数={}", method.getName(), Arrays.toString(params), ex);
        
        // 可以在这里实现异常告警、重试等逻辑
        handleAsyncException(ex, method, params);
    }
    
    /**
     * 处理异步异常
     */
    private void handleAsyncException(Throwable ex, Method method, Object... params) {
        // 实现异常处理逻辑，如发送告警、记录日志等
        AsyncManager.me().execute(() -> {
            log.info("发送异步事件异常告警: 方法={}", method.getName());
        });
    }
}

自定义事件实现

业务事件定义

/**
 * 订单事件基类
 */
public abstract class OrderEvent extends ApplicationEvent {
    
    private final Long orderId;
    private final Long userId;
    private final Date timestamp;
    
    public OrderEvent(Object source, Long orderId, Long userId) {
        super(source);
        this.orderId = orderId;
        this.userId = userId;
        this.timestamp = new Date();
    }
    
    public Long getOrderId() {
        return orderId;
    }
    
    public Long getUserId() {
        return userId;
    }
    
    public Date getTimestamp() {
        return timestamp;
    }
}

/**
 * 订单创建事件
 */
public class OrderCreatedEvent extends OrderEvent {
    
    private final BigDecimal amount;
    private final String productName;
    
    public OrderCreatedEvent(Object source, Long orderId, Long userId, 
                           BigDecimal amount, String productName) {
        super(source, orderId, userId);
        this.amount = amount;
        this.productName = productName;
    }
    
    public BigDecimal getAmount() {
        return amount;
    }
    
    public String getProductName() {
        return productName;
    }
}

/**
 * 订单支付事件
 */
public class OrderPaidEvent extends OrderEvent {
    
    private final String paymentMethod;
    private final BigDecimal paidAmount;
    
    public OrderPaidEvent(Object source, Long orderId, Long userId, 
                         String paymentMethod, BigDecimal paidAmount) {
        super(source, orderId, userId);
        this.paymentMethod = paymentMethod;
        this.paidAmount = paidAmount;
    }
    
    public String getPaymentMethod() {
        return paymentMethod;
    }
    
    public BigDecimal getPaidAmount() {
        return paidAmount;
    }
}

/**
 * 订单取消事件
 */
public class OrderCancelledEvent extends OrderEvent {
    
    private final String reason;
    private final String cancelledBy;
    
    public OrderCancelledEvent(Object source, Long orderId, Long userId, 
                             String reason, String cancelledBy) {
        super(source, orderId, userId);
        this.reason = reason;
        this.cancelledBy = cancelledBy;
    }
    
    public String getReason() {
        return reason;
    }
    
    public String getCancelledBy() {
        return cancelledBy;
    }
}

/**
 * 系统事件
 */
public class SystemEvent extends ApplicationEvent {
    
    private final String eventType;
    private final String message;
    private final Map<String, Object> data;
    private final Date timestamp;
    
    public SystemEvent(Object source, String eventType, String message, Map<String, Object> data) {
        super(source);
        this.eventType = eventType;
        this.message = message;
        this.data = data != null ? data : new HashMap<>();
        this.timestamp = new Date();
    }
    
    public String getEventType() {
        return eventType;
    }
    
    public String getMessage() {
        return message;
    }
    
    public Map<String, Object> getData() {
        return data;
    }
    
    public Date getTimestamp() {
        return timestamp;
    }
}

业务事件监听器

/**
 * 订单事件监听器
 */
@Component
@Slf4j
public class OrderEventListener {
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private NotificationService notificationService;
    
    @Autowired
    private PointsService pointsService;
    
    @Autowired
    private LogisticsService logisticsService;
    
    /**
     * 处理订单创建事件
     */
    @EventListener
    @Order(1) // 设置监听器执行顺序
    public void handleOrderCreated(OrderCreatedEvent event) {
        log.info("处理订单创建事件: 订单ID={}, 用户ID={}, 金额={}", 
                event.getOrderId(), event.getUserId(), event.getAmount());
        
        try {
            // 扣减库存
            inventoryService.reduceInventory(event.getOrderId());
            
            // 发送订单确认通知
            notificationService.sendOrderConfirmation(event.getUserId(), event.getOrderId());
            
        } catch (Exception e) {
            log.error("处理订单创建事件失败", e);
            // 可以发布订单创建失败事件
            publishOrderCreationFailedEvent(event, e);
        }
    }
    
    /**
     * 处理订单支付事件
     */
    @EventListener
    @Async("asyncEventExecutor")
    public void handleOrderPaid(OrderPaidEvent event) {
        log.info("处理订单支付事件: 订单ID={}, 支付方式={}, 支付金额={}", 
                event.getOrderId(), event.getPaymentMethod(), event.getPaidAmount());
        
        try {
            // 增加用户积分
            pointsService.addPoints(event.getUserId(), event.getPaidAmount());
            
            // 安排发货
            logisticsService.arrangeShipment(event.getOrderId());
            
            // 发送支付成功通知
            notificationService.sendPaymentSuccessNotification(event.getUserId(), event.getOrderId());
            
        } catch (Exception e) {
            log.error("处理订单支付事件失败", e);
        }
    }
    
    /**
     * 处理订单取消事件
     */
    @EventListener
    public void handleOrderCancelled(OrderCancelledEvent event) {
        log.info("处理订单取消事件: 订单ID={}, 取消原因={}, 取消人={}", 
                event.getOrderId(), event.getReason(), event.getCancelledBy());
        
        try {
            // 恢复库存
            inventoryService.restoreInventory(event.getOrderId());
            
            // 处理退款
            if ("USER".equals(event.getCancelledBy())) {
                processRefund(event.getOrderId());
            }
            
            // 发送取消通知
            notificationService.sendOrderCancellationNotification(event.getUserId(), event.getOrderId());
            
        } catch (Exception e) {
            log.error("处理订单取消事件失败", e);
        }
    }
    
    /**
     * 发布订单创建失败事件
     */
    private void publishOrderCreationFailedEvent(OrderCreatedEvent originalEvent, Exception exception) {
        // 实现订单创建失败事件发布逻辑
        log.info("发布订单创建失败事件: 订单ID={}", originalEvent.getOrderId());
    }
    
    /**
     * 处理退款
     */
    private void processRefund(Long orderId) {
        // 实现退款处理逻辑
        log.debug("处理订单退款: 订单ID={}", orderId);
    }
}

/**
 * 系统事件监听器
 */
@Component
@Slf4j
public class SystemEventListener {
    
    @Autowired
    private AlertService alertService;
    
    @Autowired
    private MetricsService metricsService;
    
    /**
     * 处理系统事件
     */
    @EventListener
    public void handleSystemEvent(SystemEvent event) {
        log.info("处理系统事件: 类型={}, 消息={}", event.getEventType(), event.getMessage());
        
        try {
            // 记录系统指标
            metricsService.recordSystemEvent(event.getEventType(), event.getData());
            
            // 根据事件类型处理
            switch (event.getEventType()) {
                case "SYSTEM_ERROR":
                    handleSystemError(event);
                    break;
                case "SYSTEM_WARNING":
                    handleSystemWarning(event);
                    break;
                case "SYSTEM_INFO":
                    handleSystemInfo(event);
                    break;
                default:
                    log.debug("未知系统事件类型: {}", event.getEventType());
            }
            
        } catch (Exception e) {
            log.error("处理系统事件失败", e);
        }
    }
    
    /**
     * 处理系统错误事件
     */
    private void handleSystemError(SystemEvent event) {
        // 发送紧急告警
        alertService.sendCriticalAlert(event.getMessage(), event.getData());
    }
    
    /**
     * 处理系统警告事件
     */
    private void handleSystemWarning(SystemEvent event) {
        // 发送警告告警
        alertService.sendWarningAlert(event.getMessage(), event.getData());
    }
    
    /**
     * 处理系统信息事件
     */
    private void handleSystemInfo(SystemEvent event) {
        // 记录信息日志
        log.info("系统信息事件: {}", event.getMessage());
    }
}

事件驱动架构

事件总线

/**
 * 事件总线
 */
@Component
@Slf4j
public class EventBus {
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    private static final String EVENT_CHANNEL = "event_channel";
    
    /**
     * 发布本地事件
     */
    public void publishLocalEvent(ApplicationEvent event) {
        try {
            eventPublisher.publishEvent(event);
            log.debug("发布本地事件: {}", event.getClass().getSimpleName());
        } catch (Exception e) {
            log.error("发布本地事件失败", e);
        }
    }
    
    /**
     * 发布分布式事件
     */
    public void publishDistributedEvent(ApplicationEvent event) {
        try {
            // 先发布本地事件
            publishLocalEvent(event);
            
            // 再发布到Redis频道
            EventMessage message = new EventMessage(event);
            redisTemplate.convertAndSend(EVENT_CHANNEL, message);
            
            log.debug("发布分布式事件: {}", event.getClass().getSimpleName());
            
        } catch (Exception e) {
            log.error("发布分布式事件失败", e);
        }
    }
    
    /**
     * 监听分布式事件
     */
    @RedisListener(EVENT_CHANNEL)
    public void handleDistributedEvent(EventMessage message) {
        try {
            ApplicationEvent event = message.getEvent();
            
            // 避免重复处理本地发布的事件
            if (!isLocalEvent(message)) {
                eventPublisher.publishEvent(event);
                log.debug("处理分布式事件: {}", event.getClass().getSimpleName());
            }
            
        } catch (Exception e) {
            log.error("处理分布式事件失败", e);
        }
    }
    
    /**
     * 判断是否为本地事件
     */
    private boolean isLocalEvent(EventMessage message) {
        // 通过消息来源判断是否为本地事件
        return message.getSource().equals(getLocalNodeId());
    }
    
    /**
     * 获取本地节点ID
     */
    private String getLocalNodeId() {
        // 实现获取本地节点ID的逻辑
        return "local-node-" + System.currentTimeMillis();
    }
}

/**
 * 事件消息包装类
 */
@Data
public class EventMessage implements Serializable {
    
    private String source;
    private String eventType;
    private ApplicationEvent event;
    private Date timestamp;
    
    public EventMessage(ApplicationEvent event) {
        this.event = event;
        this.eventType = event.getClass().getSimpleName();
        this.source = getLocalNodeId();
        this.timestamp = new Date();
    }
    
    private String getLocalNodeId() {
        // 实现获取本地节点ID的逻辑
        return "local-node-" + System.currentTimeMillis();
    }
}

/**
 * 事件存储服务
 */
@Service
@Slf4j
public class EventStoreService {
    
    @Autowired
    private EventStoreMapper eventStoreMapper;
    
    /**
     * 存储事件
     */
    public void storeEvent(ApplicationEvent event) {
        try {
            EventStore eventStore = new EventStore();
            eventStore.setEventType(event.getClass().getSimpleName());
            eventStore.setEventData(JsonUtils.toJsonString(event));
            eventStore.setTimestamp(new Date());
            eventStore.setSource(event.getSource().toString());
            
            eventStoreMapper.insert(eventStore);
            
            log.debug("存储事件: {}", event.getClass().getSimpleName());
            
        } catch (Exception e) {
            log.error("存储事件失败", e);
        }
    }
    
    /**
     * 查询事件
     */
    public List<EventStore> queryEvents(String eventType, Date startTime, Date endTime) {
        try {
            LambdaQueryWrapper<EventStore> wrapper = new LambdaQueryWrapper<>();
            
            if (StringUtils.hasText(eventType)) {
                wrapper.eq(EventStore::getEventType, eventType);
            }
            
            if (startTime != null) {
                wrapper.ge(EventStore::getTimestamp, startTime);
            }
            
            if (endTime != null) {
                wrapper.le(EventStore::getTimestamp, endTime);
            }
            
            wrapper.orderByDesc(EventStore::getTimestamp);
            
            return eventStoreMapper.selectList(wrapper);
            
        } catch (Exception e) {
            log.error("查询事件失败", e);
            return new ArrayList<>();
        }
    }
}

/**
 * 事件存储实体
 */
@Data
@TableName("event_store")
public class EventStore {
    
    @TableId(type = IdType.AUTO)
    private Long id;
    
    private String eventType;
    private String eventData;
    private String source;
    private Date timestamp;
}

总结

本文详细介绍了Spring事件机制，包括：

1. 事件发布与监听：基础事件定义、事件监听器、事件发布器
2. 异步事件处理：异步事件配置、异步监听器、异常处理
3. 自定义事件实现：业务事件定义、业务事件监听器
4. 事件驱动架构：事件总线、分布式事件、事件存储

Spring事件机制为应用提供了松耦合的模块间通信方式，是构建事件驱动架构的重要基础。

至此，第五篇Spring框架核心技术的内容就完成了。

参考资料

• Spring事件机制官方文档
• Spring异步处理
• 事件驱动架构设计模式