Spring Event 企业级应用

Spring 框架提供的事件驱动编程模型，基于观察者模式实现。它允许应用程序的不同组件之间进行松耦合的通信，当一个组件发布事件时，其他监听该事件的组件可以异步或同步地做出响应。

电商项目中的使用

①、定义事件

• BaseEvent 基础事件抽象类
• UserRegisteredEvent extends BaseEvent 用户注册事件
• OrderCreatedEvent extends BaseEvent 订单创建事件
• PaymentSuccessEvent extends BaseEvent 支付成功事件
• InventoryDeductedEvent extends BaseEvent 库存扣减事件
• SystemAlertEvent extends BaseEvent 系统告警事件

public abstract class BaseEvent{protected String eventId;protected LocalDateTime timestamp;protected String source;public BaseEvent(String source){this.eventId = UUID.randomUUID().toString();this.timestamp = LocalDateTime.now();this.source = source;}//getter...
}

// 用户注册事件
public class UserRegisteredEvent extends BaseEvent {private Long userId;private String username;private String email;private String phone;public UserRegisteredEvent(String source, Long userId, String username, String email, String phone) {super(source);this.userId = userId;this.username = username;this.email = email;this.phone = phone;}// getters...
}

// 订单创建事件
public class OrderCreatedEvent extends BaseEvent {private Long orderId;private Long userId;private BigDecimal amount;private List<OrderItem> items;public OrderCreatedEvent(String source, Long orderId, Long userId, BigDecimal amount, List<OrderItem> items) {super(source);this.orderId = orderId;this.userId = userId;this.amount = amount;this.items = items;}// getters...
}

// 支付成功事件
public class PaymentSuccessEvent extends BaseEvent {private Long paymentId;private Long orderId;private BigDecimal amount;private String paymentMethod;public PaymentSuccessEvent(String source, Long paymentId, Long orderId, BigDecimal amount, String paymentMethod) {super(source);this.paymentId = paymentId;this.orderId = orderId;this.amount = amount;this.paymentMethod = paymentMethod;}// getters...
}

// 库存扣减事件
public class InventoryDeductedEvent extends BaseEvent {private Long orderId;private List<InventoryItem> deductedItems;public InventoryDeductedEvent(String source, Long orderId, List<InventoryItem> deductedItems) {super(source);this.orderId = orderId;this.deductedItems = deductedItems;}// getters...
}

// 系统告警事件
public class SystemAlertEvent extends BaseEvent {private AlertLevel level;private String module;private String message;private Map<String, Object> context;public SystemAlertEvent(String source, AlertLevel level, String module, String message, Map<String, Object> context) {super(source);this.level = level;this.module = module;this.message = message;this.context = context;}// getters...
}public enum AlertLevel {INFO, WARNING, ERROR, CRITICAL
}

②、事件监听器

• EmailNotificationListener 邮件通知监听器
• SmsNotificationListener SMS通知监听器
• PointsEventListener 积分处理监听器
• InventoryEventListener 库存管理监听器
• AnalyticsEventListener 数据分析监听器
• MonitoringEventListener 统监控监听器
• AuditLogEventListener 审计日志监听器

@Component
@Slf4j
public class EmailNotifactionListener{@Async("emailTaskExecutor")@EventListener@Order(1) //执行顺序public void handleUserRegistered(UserRegisteredEvent event){try {log.info("发送欢迎邮件给用户: {}", event.getEmail());// 模拟邮件发送Thread.sleep(100);log.info("欢迎邮件发送成功: {}", event.getEmail());} catch (Exception e) {log.error("发送欢迎邮件失败", e);}}@Async("emailTaskExecutor")@EventListenerpublic void handleOrderCreated(OrderCreatedEvent event) {log.info("发送订单确认邮件, 订单ID: {}", event.getOrderId());// 实际邮件发送逻辑...}@Async("emailTaskExecutor")@EventListenerpublic void handlePaymentSuccess(PaymentSuccessEvent event) {log.info("发送支付成功邮件, 订单ID: {}", event.getOrderId());// 实际邮件发送逻辑...}
}

// SMS通知监听器
@Component
@Slf4j
public class SmsNotificationListener {@Async("smsTaskExecutor")@EventListenerpublic void handleUserRegistered(UserRegisteredEvent event) {if (StringUtils.isNotBlank(event.getPhone())) {log.info("发送欢迎短信到: {}", event.getPhone());// 实际短信发送逻辑...}}@Async("smsTaskExecutor")@EventListenerpublic void handlePaymentSuccess(PaymentSuccessEvent event) {log.info("发送支付成功短信, 订单ID: {}", event.getOrderId());// 实际短信发送逻辑...}
}

// 积分处理监听器
@Component
@Slf4j
public class PointsEventListener {@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)public void handleUserRegistration(UserRegisteredEvent event) {log.info("为新用户赠送积分, 用户ID: {}", event.getUserId());// 积分服务调用...}@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)public void handleOrderPayment(PaymentSuccessEvent event) {log.info("为订单支付增加积分, 订单ID: {}", event.getOrderId());// 根据订单金额计算积分...}
}

// 库存管理监听器
@Component
@Slf4j
public class InventoryEventListener {@TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)public void handleOrderCreated(OrderCreatedEvent event) {log.info("预扣减库存, 订单ID: {}", event.getOrderId());// 库存预扣减逻辑...// 发布库存扣减事件ApplicationEventPublisher eventPublisher = ...;eventPublisher.publishEvent(new InventoryDeductedEvent("InventoryService", event.getOrderId(), deductedItems));}@EventListenerpublic void handleInventoryDeducted(InventoryDeductedEvent event) {log.info("库存扣减完成, 订单ID: {}", event.getOrderId());// 更新库存状态...}
}

// 数据分析监听器
@Component
@Slf4j
public class AnalyticsEventListener {@Async("analyticsTaskExecutor")@EventListenerpublic void handleUserBehavior(BaseEvent event) {log.info("记录用户行为事件到数据分析系统: {}", event.getClass().getSimpleName());// 发送到Kafka或写入数据仓库...}
}

// 系统监控监听器
@Component
@Slf4j
public class MonitoringEventListener {@EventListenerpublic void handleSystemAlert(SystemAlertEvent event) {switch (event.getLevel()) {case ERROR:case CRITICAL:log.error("系统告警: {} - {} - {}", event.getModule(), event.getMessage(), event.getContext());// 发送到监控平台...break;case WARNING:log.warn("系统警告: {} - {}", event.getModule(), event.getMessage());break;case INFO:log.info("系统信息: {} - {}", event.getModule(), event.getMessage());break;}}
}

// 审计日志监听器
@Component
@Slf4j
public class AuditLogEventListener {@EventListenerpublic void handleAllEvents(BaseEvent event) {log.info("审计日志 - 事件类型: {}, 事件ID: {}, 时间: {}", event.getClass().getSimpleName(), event.getEventId(), event.getTimestamp());// 写入审计数据库auditLogService.save(new AuditLog(event.getEventId(),event.getClass().getSimpleName(),event.getSource(),event.getTimestamp(),event));}
}

③、事件发布服务

@Service
@Slf4j
public class EventPublisherService {private final ApplicationEventPublisher eventPublisher;private final MeterRegistry meterRegistry;public EventPublisherService(ApplicationEventPublisher eventPublisher, MeterRegistry meterRegistry) {this.eventPublisher = eventPublisher;this.meterRegistry = meterRegistry;}public void publishUserRegistered(Long userId, String username, String email, String phone) {UserRegisteredEvent event = new UserRegisteredEvent("UserService", userId, username, email, phone);publishEvent(event);}public void publishOrderCreated(Long orderId, Long userId, BigDecimal amount, List<OrderItem> items) {OrderCreatedEvent event = new OrderCreatedEvent("OrderService", orderId, userId, amount, items);publishEvent(event);}public void publishPaymentSuccess(Long paymentId, Long orderId, BigDecimal amount, String paymentMethod) {PaymentSuccessEvent event = new PaymentSuccessEvent("PaymentService", paymentId, orderId, amount, paymentMethod);publishEvent(event);}public void publishSystemAlert(AlertLevel level, String module, String message, Map<String, Object> context) {SystemAlertEvent event = new SystemAlertEvent(module, level, module, message, context);publishEvent(event);}private void publishEvent(BaseEvent event) {try {eventPublisher.publishEvent(event);meterRegistry.counter("event.published", "type", event.getClass().getSimpleName()).increment();log.debug("事件发布成功: {}", event.getEventId());} catch (Exception e) {log.error("事件发布失败: {}", event.getEventId(), e);meterRegistry.counter("event.publish.failed", "type", event.getClass().getSimpleName()).increment();}}
}

④、配置类

@Configuration
@EnableAsync
@EnableScheduling
public class EventConfig {// 邮件任务线程池@Bean("emailTaskExecutor")public TaskExecutor emailTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(5);executor.setMaxPoolSize(10);executor.setQueueCapacity(100);executor.setThreadNamePrefix("email-exec-");executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());executor.initialize();return executor;}// SMS任务线程池@Bean("smsTaskExecutor")public TaskExecutor smsTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(3);executor.setMaxPoolSize(5);executor.setQueueCapacity(50);executor.setThreadNamePrefix("sms-exec-");executor.initialize();return executor;}// 数据分析任务线程池@Bean("analyticsTaskExecutor")public TaskExecutor analyticsTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(10);executor.setMaxPoolSize(20);executor.setQueueCapacity(1000);executor.setThreadNamePrefix("analytics-exec-");executor.initialize();return executor;}
}

// 事件监控配置
@Configuration
public class EventMonitoringConfig {@Beanpublic MeterRegistryCustomizer<MeterRegistry> eventMetrics() {return registry -> {// 事件发布统计Counter.builder("event.published").description("已发布事件数量").tag("type", "all").register(registry);// 事件处理时间统计Timer.builder("event.processing.time").description("事件处理时间").register(registry);};}
}

⑤、业务服务中使用

@Service
@Slf4j
public class UserService {private final UserRepository userRepository;private final EventPublisherService eventPublisher;public UserService(UserRepository userRepository, EventPublisherService eventPublisher) {this.userRepository = userRepository;this.eventPublisher = eventPublisher;}@Transactionalpublic User registerUser(UserRegistrationRequest request) {// 验证业务逻辑...User user = new User();user.setUsername(request.getUsername());user.setEmail(request.getEmail());user.setPhone(request.getPhone());User savedUser = userRepository.save(user);// 发布用户注册事件eventPublisher.publishUserRegistered(savedUser.getId(),savedUser.getUsername(),savedUser.getEmail(),savedUser.getPhone());return savedUser;}
}@Service
@Slf4j
public class OrderService {private final OrderRepository orderRepository;private final EventPublisherService eventPublisher;public OrderService(OrderRepository orderRepository, EventPublisherService eventPublisher) {this.orderRepository = orderRepository;this.eventPublisher = eventPublisher;}@Transactionalpublic Order createOrder(OrderCreateRequest request) {// 创建订单逻辑...Order order = new Order();order.setUserId(request.getUserId());order.setAmount(calculateAmount(request.getItems()));order.setItems(request.getItems());Order savedOrder = orderRepository.save(order);// 发布订单创建事件eventPublisher.publishOrderCreated(savedOrder.getId(),savedOrder.getUserId(),savedOrder.getAmount(),savedOrder.getItems());return savedOrder;}
}

@Service
@Slf4j
public class PaymentService {private final PaymentRepository paymentRepository;private final EventPublisherService eventPublisher;public PaymentService(PaymentRepository paymentRepository, EventPublisherService eventPublisher) {this.paymentRepository = paymentRepository;this.eventPublisher = eventPublisher;}@Transactionalpublic Payment processPayment(PaymentRequest request) {// 支付处理逻辑...Payment payment = new Payment();payment.setOrderId(request.getOrderId());payment.setAmount(request.getAmount());payment.setPaymentMethod(request.getPaymentMethod());payment.setStatus(PaymentStatus.SUCCESS);Payment savedPayment = paymentRepository.save(payment);// 发布支付成功事件eventPublisher.publishPaymentSuccess(savedPayment.getId(),savedPayment.getOrderId(),savedPayment.getAmount(),savedPayment.getPaymentMethod());return savedPayment;}
}

⑥、监控和健康检查

@Component
@Slf4j
public class EventHealthIndicator implements HealthIndicator {private final MeterRegistry meterRegistry;public EventHealthIndicator(MeterRegistry meterRegistry) {this.meterRegistry = meterRegistry;}@Overridepublic Health health() {double failedEvents = meterRegistry.counter("event.publish.failed", "type", "all").count();double totalEvents = meterRegistry.counter("event.published", "type", "all").count();double failureRate = totalEvents > 0 ? failedEvents / totalEvents : 0;Health.Builder builder = failureRate > 0.1 ? Health.down() : Health.up();return builder.withDetail("total_events", totalEvents).withDetail("failed_events", failedEvents).withDetail("failure_rate", String.format("%.2f%%", failureRate * 100)).build();}
}

// 事件统计端点
@RestController
@RequestMapping("/admin/events")
public class EventStatisticsController {private final MeterRegistry meterRegistry;public EventStatisticsController(MeterRegistry meterRegistry) {this.meterRegistry = meterRegistry;}@GetMapping("/statistics")public Map<String, Object> getEventStatistics() {Map<String, Object> stats = new HashMap<>();// 获取各种事件类型的统计meterRegistry.find("event.published").counters().forEach(counter -> {String type = counter.getId().getTag("type");stats.put("published_" + type, counter.count());});meterRegistry.find("event.publish.failed").counters().forEach(counter -> {String type = counter.getId().getTag("type");stats.put("failed_" + type, counter.count());});return stats;}
}