// 事件基类 - 定义事件的基本结构
@Data
@AllArgsConstructor
@NoArgsConstructor
public abstract class DomainEvent {
    /**
     * 事件ID
     */
    private String eventId;
    
    /**
     * 事件发生时间
     */
    private LocalDateTime eventTime;
    
    /**
     * 事件源
     */
    private String source;
    
    /**
     * 事件版本
     */
    private String version;
    
    protected DomainEvent(String source) {
        this.eventId = UUID.randomUUID().toString();
        this.eventTime = LocalDateTime.now();
        this.source = source;
        this.version = "1.0";
    }
}

// 订单状态变更事件
@Data
@EqualsAndHashCode(callSuper = true)
public class OrderStatusChangedEvent extends DomainEvent {
    /**
     * 订单ID
     */
    private String orderId;
    
    /**
     * 订单号
     */
    private String orderNo;
    
    /**
     * 原状态
     */
    private String oldStatus;
    
    /**
     * 新状态
     */
    private String newStatus;
    
    /**
     * 用户ID
     */
    private String userId;
    
    /**
     * 订单金额
     */
    private BigDecimal amount;
    
    /**
     * 扩展信息
     */
    private Map<String, Object> extensions;
    
    public OrderStatusChangedEvent(String orderId, String orderNo, 
                                  String oldStatus, String newStatus,
                                  String userId, BigDecimal amount) {
        super("OrderService");
        this.orderId = orderId;
        this.orderNo = orderNo;
        this.oldStatus = oldStatus;
        this.newStatus = newStatus;
        this.userId = userId;
        this.amount = amount;
        this.extensions = new HashMap<>();
    }
}

// 观察者接口
public interface DomainEventObserver<T extends DomainEvent> {
    /**
     * 处理事件
     * @param event 事件对象
     */
    void handle(T event);
    
    /**
     * 获取观察者名称
     */
    String getObserverName();
    
    /**
     * 获取支持的事件类型
     */
    Class<T> getEventType();
    
    /**
     * 获取优先级（数值越小优先级越高）
     */
    default int getPriority() {
        return 100;
    }
    
    /**
     * 是否支持异步处理
     */
    default boolean supportAsync() {
        return true;
    }
}

// 被观察者接口
public interface DomainEventPublisher {
    /**
     * 注册观察者
     */
    void register(DomainEventObserver<? extends DomainEvent> observer);
    
    /**
     * 移除观察者
     */
    void unregister(DomainEventObserver<? extends DomainEvent> observer);
    
    /**
     * 发布事件
     */
    void publish(DomainEvent event);
    
    /**
     * 异步发布事件
     */
    void publishAsync(DomainEvent event);
    
    /**
     * 获取已注册的观察者列表
     */
    List<DomainEventObserver<? extends DomainEvent>> getObservers();
}

// 事件发布器实现
@Component
@Slf4j
public class DefaultDomainEventPublisher implements DomainEventPublisher {
    
    private final Map<Class<? extends DomainEvent>, List<DomainEventObserver<? extends DomainEvent>>> observerMap;
    private final ExecutorService asyncExecutor;
    private final ApplicationEventPublisher applicationEventPublisher;
    
    public DefaultDomainEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        this.observerMap = new ConcurrentHashMap<>();
        this.asyncExecutor = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors() * 2,
            new ThreadFactoryBuilder()
                .setNameFormat("event-publisher-%d")
                .setDaemon(true)
                .build()
        );
        this.applicationEventPublisher = applicationEventPublisher;
    }
    
    @Override
    public void register(DomainEventObserver<? extends DomainEvent> observer) {
        Class<? extends DomainEvent> eventType = observer.getEventType();
        observerMap.computeIfAbsent(eventType, k -> new ArrayList<>()).add(observer);
        
        // 按优先级排序
        observerMap.get(eventType).sort(Comparator.comparingInt(DomainEventObserver::getPriority));
        
        log.info("注册观察者成功: eventType={}, observerName={}, priority={}", 
                eventType.getSimpleName(), observer.getObserverName(), observer.getPriority());
    }
    
    @Override
    public void unregister(DomainEventObserver<? extends DomainEvent> observer) {
        Class<? extends DomainEvent> eventType = observer.getEventType();
        List<DomainEventObserver<? extends DomainEvent>> observers = observerMap.get(eventType);
        if (observers != null) {
            observers.remove(observer);
            log.info("移除观察者成功: eventType={}, observerName={}", 
                    eventType.getSimpleName(), observer.getObserverName());
        }
    }
    
    @Override
    public void publish(DomainEvent event) {
        if (event == null) {
            return;
        }
        
        List<DomainEventObserver<? extends DomainEvent>> observers = observerMap.get(event.getClass());
        if (observers == null || observers.isEmpty()) {
            log.debug("没有找到对应的观察者: eventType={}", event.getClass().getSimpleName());
            return;
        }
        
        log.info("开始发布事件: eventId={}, eventType={}, observerCount={}", 
                event.getEventId(), event.getClass().getSimpleName(), observers.size());
        
        for (DomainEventObserver<? extends DomainEvent> observer : observers) {
            try {
                // 同步处理
                observer.handle(event);
                log.debug("观察者处理事件成功: observerName={}, eventId={}", 
                         observer.getObserverName(), event.getEventId());
            } catch (Exception e) {
                log.error("观察者处理事件失败: observerName={}, eventId={}", 
                         observer.getObserverName(), event.getEventId(), e);
                // 根据业务需要决定是否继续处理其他观察者
                throw new EventHandleException("事件处理失败: " + observer.getObserverName(), e);
            }
        }
        
        // 发布Spring事件，支持其他监听器
        applicationEventPublisher.publishEvent(event);
        
        log.info("事件发布完成: eventId={}", event.getEventId());
    }
    
    @Override
    public void publishAsync(DomainEvent event) {
        if (event == null) {
            return;
        }
        
        List<DomainEventObserver<? extends DomainEvent>> observers = observerMap.get(event.getClass());
        if (observers == null || observers.isEmpty()) {
            log.debug("没有找到对应的观察者: eventType={}", event.getClass().getSimpleName());
            return;
        }
        
        log.info("开始异步发布事件: eventId={}, eventType={}, observerCount={}", 
                event.getEventId(), event.getClass().getSimpleName(), observers.size());
        
        for (DomainEventObserver<? extends DomainEvent> observer : observers) {
            if (observer.supportAsync()) {
                asyncExecutor.submit(() -> {
                    try {
                        observer.handle(event);
                        log.debug("异步观察者处理事件成功: observerName={}, eventId={}", 
                                 observer.getObserverName(), event.getEventId());
                    } catch (Exception e) {
                        log.error("异步观察者处理事件失败: observerName={}, eventId={}", 
                                 observer.getObserverName(), event.getEventId(), e);
                    }
                });
            } else {
                // 不支持异步的观察者，同步处理
                try {
                    observer.handle(event);
                    log.debug("同步观察者处理事件成功: observerName={}, eventId={}", 
                             observer.getObserverName(), event.getEventId());
                } catch (Exception e) {
                    log.error("同步观察者处理事件失败: observerName={}, eventId={}", 
                             observer.getObserverName(), event.getEventId(), e);
                }
            }
        }
        
        // 异步发布Spring事件
        asyncExecutor.submit(() -> applicationEventPublisher.publishEvent(event));
        
        log.info("异步事件发布完成: eventId={}", event.getEventId());
    }
    
    @Override
    public List<DomainEventObserver<? extends DomainEvent>> getObservers() {
        return observerMap.values().stream()
                .flatMap(List::stream)
                .collect(Collectors.toList());
    }
    
    @PreDestroy
    public void destroy() {
        if (asyncExecutor != null && !asyncExecutor.isShutdown()) {
            asyncExecutor.shutdown();
            try {
                if (!asyncExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
                    asyncExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                asyncExecutor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
}

// 具体观察者 - 库存系统
@Component
@Slf4j
public class InventoryObserver implements DomainEventObserver<OrderStatusChangedEvent> {
    
    private final InventoryService inventoryService;
    
    public InventoryObserver(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }
    
    @Override
    public void handle(OrderStatusChangedEvent event) {
        log.info("库存系统处理订单状态变更事件: orderId={}, newStatus={}", 
                event.getOrderId(), event.getNewStatus());
        
        // 只有订单支付成功时才扣减库存
        if ("PAID".equals(event.getNewStatus())) {
            try {
                inventoryService.deductInventory(event.getOrderId());
                log.info("库存扣减成功: orderId={}", event.getOrderId());
            } catch (Exception e) {
                log.error("库存扣减失败: orderId={}", event.getOrderId(), e);
                throw new RuntimeException("库存扣减失败", e);
            }
        }
        
        // 订单取消时恢复库存
        if ("CANCELLED".equals(event.getNewStatus()) && "PAID".equals(event.getOldStatus())) {
            try {
                inventoryService.restoreInventory(event.getOrderId());
                log.info("库存恢复成功: orderId={}", event.getOrderId());
            } catch (Exception e) {
                log.error("库存恢复失败: orderId={}", event.getOrderId(), e);
                throw new RuntimeException("库存恢复失败", e);
            }
        }
    }
    
    @Override
    public String getObserverName() {
        return "InventoryObserver";
    }
    
    @Override
    public Class<OrderStatusChangedEvent> getEventType() {
        return OrderStatusChangedEvent.class;
    }
    
    @Override
    public int getPriority() {
        return 10; // 高优先级，库存操作很重要
    }
    
    @Override
    public boolean supportAsync() {
        return false; // 库存操作需要同步执行，保证数据一致性
    }
}

// 具体观察者 - 消息推送系统
@Component
@Slf4j
public class NotificationObserver implements DomainEventObserver<OrderStatusChangedEvent> {
    
    private final NotificationService notificationService;
    private final UserService userService;
    
    public NotificationObserver(NotificationService notificationService, UserService userService) {
        this.notificationService = notificationService;
        this.userService = userService;
    }
    
    @Override
    public void handle(OrderStatusChangedEvent event) {
        log.info("消息推送系统处理订单状态变更事件: orderId={}, newStatus={}", 
                event.getOrderId(), event.getNewStatus());
        
        try {
            // 获取用户信息
            User user = userService.getUserById(event.getUserId());
            if (user == null) {
                log.warn("用户不存在: userId={}", event.getUserId());
                return;
            }
            
            // 构建推送消息
            String message = buildNotificationMessage(event);
            
            // 发送推送
            notificationService.sendPush(user.getDeviceToken(), message);
            
            log.info("消息推送成功: orderId={}, userId={}", event.getOrderId(), event.getUserId());
            
        } catch (Exception e) {
            log.error("消息推送失败: orderId={}, userId={}", 
                     event.getOrderId(), event.getUserId(), e);
            // 推送失败不影响主流程，只记录日志
        }
    }
    
    private String buildNotificationMessage(OrderStatusChangedEvent event) {
        switch (event.getNewStatus()) {
            case "PAID":
                return String.format("您的订单 %s 已支付成功，正在准备发货", event.getOrderNo());
            case "SHIPPED":
                return String.format("您的订单 %s 已发货，请注意查收", event.getOrderNo());
            case "DELIVERED":
                return String.format("您的订单 %s 已送达，感谢您的购买", event.getOrderNo());
            case "CANCELLED":
                return String.format("您的订单 %s 已取消，如有疑问请联系客服", event.getOrderNo());
            default:
                return String.format("您的订单 %s 状态已更新为: %s", 
                                   event.getOrderNo(), event.getNewStatus());
        }
    }
    
    @Override
    public String getObserverName() {
        return "NotificationObserver";
    }
    
    @Override
    public Class<OrderStatusChangedEvent> getEventType() {
        return OrderStatusChangedEvent.class;
    }
    
    @Override
    public int getPriority() {
        return 50; // 中等优先级
    }
    
    @Override
    public boolean supportAsync() {
        return true; // 推送可以异步执行
    }
}

// 具体观察者 - 数据分析系统
@Component
@Slf4j
public class AnalyticsObserver implements DomainEventObserver<OrderStatusChangedEvent> {
    
    private final AnalyticsService analyticsService;
    
    public AnalyticsObserver(AnalyticsService analyticsService) {
        this.analyticsService = analyticsService;
    }
    
    @Override
    public void handle(OrderStatusChangedEvent event) {
        log.info("数据分析系统处理订单状态变更事件: orderId={}, newStatus={}", 
                event.getOrderId(), event.getNewStatus());
        
        try {
            // 记录状态变更数据
            analyticsService.recordOrderStatusChange(
                event.getOrderId(),
                event.getOrderNo(),
                event.getOldStatus(),
                event.getNewStatus(),
                event.getAmount(),
                event.getEventTime()
            );
            
            log.info("数据分析记录成功: orderId={}", event.getOrderId());
            
        } catch (Exception e) {
            log.error("数据分析记录失败: orderId={}", event.getOrderId(), e);
            // 数据分析失败不影响主流程
        }
    }
    
    @Override
    public String getObserverName() {
        return "AnalyticsObserver";
    }
    
    @Override
    public Class<OrderStatusChangedEvent> getEventType() {
        return OrderStatusChangedEvent.class;
    }
    
    @Override
    public int getPriority() {
        return 90; // 低优先级
    }
    
    @Override
    public boolean supportAsync() {
        return true; // 数据分析可以异步执行
    }
}

// 传统硬编码方式 - 问题代码示例
@Service
public class OrderService {
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private NotificationService notificationService;
    
    @Autowired
    private AnalyticsService analyticsService;
    
    @Autowired
    private LogisticsService logisticsService;
    
    @Autowired
    private FinanceService financeService;
    
    /**
     * 更新订单状态 - 硬编码调用方式
     */
    @Transactional
    public void updateOrderStatus(String orderId, String newStatus) {
        // 1. 更新数据库
        Order order = orderRepository.findById(orderId);
        String oldStatus = order.getStatus();
        order.setStatus(newStatus);
        orderRepository.save(order);
        
        // 2. 硬编码调用各个下游系统
        try {
            // 库存系统
            if ("PAID".equals(newStatus)) {
                inventoryService.deductInventory(orderId);
            }
            
            // 消息推送
            User user = userService.getUserById(order.getUserId());
            String message = buildMessage(order, newStatus);
            notificationService.sendPush(user.getDeviceToken(), message);
            
            // 数据分析
            analyticsService.recordOrderStatusChange(orderId, oldStatus, newStatus, order.getAmount());
            
            // 物流系统
            if ("SHIPPED".equals(newStatus)) {
                logisticsService.createShipment(orderId);
            }
            
            // 财务系统
            if ("PAID".equals(newStatus)) {
                financeService.recordPayment(orderId, order.getAmount());
            }
            
        } catch (Exception e) {
            log.error("通知下游系统失败: orderId={}", orderId, e);
            // 这里处理很复杂，某个系统失败是否影响其他系统？
        }
    }
}

// 订单服务 - 使用观察者模式
@Service
@Slf4j
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private DomainEventPublisher eventPublisher;
    
    /**
     * 更新订单状态 - 观察者模式实现
     */
    @Transactional
    public void updateOrderStatus(String orderId, String newStatus) {
        log.info("开始更新订单状态: orderId={}, newStatus={}", orderId, newStatus);
        
        // 1. 更新数据库
        Order order = orderRepository.findById(orderId);
        String oldStatus = order.getStatus();
        order.setStatus(newStatus);
        orderRepository.save(order);
        
        // 2. 发布事件
        OrderStatusChangedEvent event = new OrderStatusChangedEvent(
            orderId, order.getOrderNo(), oldStatus, newStatus,
            order.getUserId(), order.getAmount()
        );
        
        // 异步发布事件，不阻塞主流程
        eventPublisher.publishAsync(event);
        
        log.info("订单状态更新完成: orderId={}, eventId={}", orderId, event.getEventId());
    }
    
    /**
     * 批量更新订单状态
     */
    @Transactional
    public void batchUpdateOrderStatus(List<String> orderIds, String newStatus) {
        log.info("开始批量更新订单状态: orderCount={}, newStatus={}", orderIds.size(), newStatus);
        
        List<OrderStatusChangedEvent> events = new ArrayList<>();
        
        for (String orderId : orderIds) {
            Order order = orderRepository.findById(orderId);
            String oldStatus = order.getStatus();
            order.setStatus(newStatus);
            orderRepository.save(order);
            
            OrderStatusChangedEvent event = new OrderStatusChangedEvent(
                orderId, order.getOrderNo(), oldStatus, newStatus,
                order.getUserId(), order.getAmount()
            );
            events.add(event);
        }
        
        // 批量发布事件
        for (OrderStatusChangedEvent event : events) {
            eventPublisher.publishAsync(event);
        }
        
        log.info("批量更新订单状态完成: orderCount={}", orderIds.size());
    }
}

// 观察者自动注册配置
@Configuration
@Slf4j
public class ObserverAutoConfiguration {
    
    @Autowired
    private DomainEventPublisher eventPublisher;
    
    @Autowired
    private ApplicationContext applicationContext;
    
    @PostConstruct
    public void registerObservers() {
        // 自动注册所有实现了DomainEventObserver接口的Bean
        Map<String, DomainEventObserver> observers = applicationContext.getBeansOfType(DomainEventObserver.class);
        
        for (DomainEventObserver observer : observers.values()) {
            eventPublisher.register(observer);
            log.info("自动注册观察者: {}", observer.getObserverName());
        }
        
        log.info("观察者注册完成，共注册{}个观察者", observers.size());
    }
}

// 事件过滤器接口
public interface EventFilter {
    /**
     * 是否处理该事件
     */
    boolean shouldHandle(DomainEvent event);
    
    /**
     * 过滤器名称
     */
    String getFilterName();
}

// 基于状态的事件过滤器
@Component
public class StatusEventFilter implements EventFilter {
    
    private final Set<String> allowedStatuses;
    
    public StatusEventFilter() {
        // 只处理这些状态变更事件
        this.allowedStatuses = Set.of("PAID", "SHIPPED", "DELIVERED", "CANCELLED");
    }
    
    @Override
    public boolean shouldHandle(DomainEvent event) {
        if (event instanceof OrderStatusChangedEvent) {
            OrderStatusChangedEvent orderEvent = (OrderStatusChangedEvent) event;
            return allowedStatuses.contains(orderEvent.getNewStatus());
        }
        return true; // 其他事件默认处理
    }
    
    @Override
    public String getFilterName() {
        return "StatusEventFilter";
    }
}

// 基于金额的事件过滤器
@Component
public class AmountEventFilter implements EventFilter {
    
    private final BigDecimal thresholdAmount;
    
    public AmountEventFilter(@Value("${event.filter.amount.threshold:1000}") BigDecimal thresholdAmount) {
        this.thresholdAmount = thresholdAmount;
    }
    
    @Override
    public boolean shouldHandle(DomainEvent event) {
        if (event instanceof OrderStatusChangedEvent) {
            OrderStatusChangedEvent orderEvent = (OrderStatusChangedEvent) event;
            // 只处理金额超过阈值的订单
            return orderEvent.getAmount().compareTo(thresholdAmount) >= 0;
        }
        return true;
    }
    
    @Override
    public String getFilterName() {
        return "AmountEventFilter";
    }
}

// 带过滤器的事件发布器
@Component
@Slf4j
public class FilteredDomainEventPublisher extends DefaultDomainEventPublisher {
    
    private final List<EventFilter> eventFilters;
    
    public FilteredDomainEventPublisher(ApplicationEventPublisher applicationEventPublisher,
                                       List<EventFilter> eventFilters) {
        super(applicationEventPublisher);
        this.eventFilters = eventFilters;
    }
    
    @Override
    public void publish(DomainEvent event) {
        // 应用过滤器
        for (EventFilter filter : eventFilters) {
            if (!filter.shouldHandle(event)) {
                log.debug("事件被过滤器拒绝: eventId={}, filter={}", 
                         event.getEventId(), filter.getFilterName());
                return;
            }
        }
        
        // 调用父类方法
        super.publish(event);
    }
    
    @Override
    public void publishAsync(DomainEvent event) {
        // 应用过滤器
        for (EventFilter filter : eventFilters) {
            if (!filter.shouldHandle(event)) {
                log.debug("异步事件被过滤器拒绝: eventId={}, filter={}", 
                         event.getEventId(), filter.getFilterName());
                return;
            }
        }
        
        // 调用父类方法
        super.publishAsync(event);
    }
}

// 事件重试策略
public interface EventRetryStrategy {
    /**
     * 是否应该重试
     */
    boolean shouldRetry(DomainEvent event, Exception exception, int retryCount);
    
    /**
     * 获取重试延迟时间（毫秒）
     */
    long getRetryDelay(int retryCount);
    
    /**
     * 最大重试次数
     */
    int getMaxRetryCount();
}

// 指数退避重试策略
@Component
public class ExponentialBackoffRetryStrategy implements EventRetryStrategy {
    
    @Value("${event.retry.max.count:3}")
    private int maxRetryCount;
    
    @Value("${event.retry.base.delay:1000}")
    private long baseDelay;
    
    @Value("${event.retry.max.delay:30000}")
    private long maxDelay;
    
    @Override
    public boolean shouldRetry(DomainEvent event, Exception exception, int retryCount) {
        if (retryCount >= maxRetryCount) {
            return false;
        }
        
        // 某些异常不重试
        if (exception instanceof IllegalArgumentException) {
            return false;
        }
        
        return true;
    }
    
    @Override
    public long getRetryDelay(int retryCount) {
        long delay = baseDelay * (long) Math.pow(2, retryCount);
        return Math.min(delay, maxDelay);
    }
    
    @Override
    public int getMaxRetryCount() {
        return maxRetryCount;
    }
}

// 带重试功能的事件发布器
@Component
@Slf4j
public class RetryableDomainEventPublisher extends DefaultDomainEventPublisher {
    
    private final EventRetryStrategy retryStrategy;
    private final ScheduledExecutorService retryExecutor;
    
    public RetryableDomainEventPublisher(ApplicationEventPublisher applicationEventPublisher,
                                        EventRetryStrategy retryStrategy) {
        super(applicationEventPublisher);
        this.retryStrategy = retryStrategy;
        this.retryExecutor = Executors.newScheduledThreadPool(
            Runtime.getRuntime().availableProcessors(),
            new ThreadFactoryBuilder()
                .setNameFormat("event-retry-%d")
                .setDaemon(true)
                .build()
        );
    }
    
    @Override
    public void publish(DomainEvent event) {
        publishWithRetry(event, 0);
    }
    
    private void publishWithRetry(DomainEvent event, int retryCount) {
        try {
            super.publish(event);
        } catch (Exception e) {
            if (retryStrategy.shouldRetry(event, e, retryCount)) {
                long delay = retryStrategy.getRetryDelay(retryCount);
                log.warn("事件发布失败，准备重试: eventId={}, retryCount={}, delay={}ms", 
                        event.getEventId(), retryCount, delay, e);
                
                retryExecutor.schedule(() -> {
                    try {
                        publishWithRetry(event, retryCount + 1);
                    } catch (Exception retryException) {
                        log.error("事件重试失败: eventId={}, retryCount={}", 
                                 event.getEventId(), retryCount, retryException);
                    }
                }, delay, TimeUnit.MILLISECONDS);
            } else {
                log.error("事件发布最终失败: eventId={}, retryCount={}", 
                         event.getEventId(), retryCount, e);
                throw new EventPublishException("事件发布失败，已达到最大重试次数", e);
            }
        }
    }
    
    @PreDestroy
    public void destroy() {
        if (retryExecutor != null && !retryExecutor.isShutdown()) {
            retryExecutor.shutdown();
            try {
                if (!retryExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
                    retryExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                retryExecutor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
}

// 错误示例：循环依赖
@Component
public class ObserverA implements DomainEventObserver<OrderStatusChangedEvent> {
    @Autowired
    private DomainEventPublisher eventPublisher;
    
    @Override
    public void handle(OrderStatusChangedEvent event) {
        // 处理订单事件
        // ...
        
        // 发布用户事件，可能触发ObserverB
        UserStatusChangedEvent userEvent = new UserStatusChangedEvent(...);
        eventPublisher.publish(userEvent); // 可能触发循环
    }
}

@Component
public class ObserverB implements DomainEventObserver<UserStatusChangedEvent> {
    @Autowired
    private DomainEventPublisher eventPublisher;
    
    @Override
    public void handle(UserStatusChangedEvent event) {
        // 处理用户事件
        // ...
        
        // 发布订单事件，可能触发ObserverA
        OrderStatusChangedEvent orderEvent = new OrderStatusChangedEvent(...);
        eventPublisher.publish(orderEvent); // 形成循环
    }
}

// 带循环检测的事件发布器
@Component
@Slf4j
public class CircularDetectionEventPublisher extends DefaultDomainEventPublisher {
    
    private final ThreadLocal<Set<String>> processingEvents = ThreadLocal.withInitial(HashSet::new);
    
    @Override
    public void publish(DomainEvent event) {
        String eventKey = event.getClass().getSimpleName() + ":" + event.getEventId();
        
        // 检查循环依赖
        if (processingEvents.get().contains(eventKey)) {
            log.warn("检测到事件循环依赖，跳过处理: eventKey={}", eventKey);
            return;
        }
        
        try {
            processingEvents.get().add(eventKey);
            super.publish(event);
        } finally {
            processingEvents.get().remove(eventKey);
        }
    }
    
    @Override
    public void publishAsync(DomainEvent event) {
        String eventKey = event.getClass().getSimpleName() + ":" + event.getEventId();
        
        // 异步事件也需要检查循环
        if (processingEvents.get().contains(eventKey)) {
            log.warn("检测到异步事件循环依赖，跳过处理: eventKey={}", eventKey);
            return;
        }
        
        try {
            processingEvents.get().add(eventKey);
            super.publishAsync(event);
        } finally {
            processingEvents.get().remove(eventKey);
        }
    }
}

// 问题：库存扣减应该在消息推送之前执行
// 但由于优先级设置不当，可能先执行消息推送

// 明确设置观察者优先级
@Component
public class InventoryObserver implements DomainEventObserver<OrderStatusChangedEvent> {
    @Override
    public int getPriority() {
        return 10; // 高优先级，先执行
    }
}

@Component
public class NotificationObserver implements DomainEventObserver<OrderStatusChangedEvent> {
    @Override
    public int getPriority() {
        return 50; // 中等优先级，后执行
    }
}

// 在事件发布器中按优先级排序
@Override
public void register(DomainEventObserver<? extends DomainEvent> observer) {
    Class<? extends DomainEvent> eventType = observer.getEventType();
    observerMap.computeIfAbsent(eventType, k -> new ArrayList<>()).add(observer);
    
    // 按优先级排序（数值越小优先级越高）
    observerMap.get(eventType).sort(Comparator.comparingInt(DomainEventObserver::getPriority));
}

// 事务性事件发布
@Component
@Slf4j
public class TransactionalEventPublisher {
    
    @Autowired
    private DomainEventPublisher eventPublisher;
    
    private final ThreadLocal<List<DomainEvent>> pendingEvents = ThreadLocal.withInitial(ArrayList::new);
    
    /**
     * 在事务中发布事件（暂存）
     */
    public void publishInTransaction(DomainEvent event) {
        pendingEvents.get().add(event);
        log.debug("事件暂存，等待事务提交: eventId={}", event.getEventId());
    }
    
    /**
     * 事务提交后发布事件
     */
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void publishAfterCommit(ApplicationEvent event) {
        List<DomainEvent> events = pendingEvents.get();
        if (!events.isEmpty()) {
            log.info("事务提交后发布事件: eventCount={}", events.size());
            for (DomainEvent domainEvent : events) {
                eventPublisher.publishAsync(domainEvent);
            }
            pendingEvents.get().clear();
        }
    }
    
    /**
     * 事务回滚时清除事件
     */
    @TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
    public void clearAfterRollback(ApplicationEvent event) {
        List<DomainEvent> events = pendingEvents.get();
        if (!events.isEmpty()) {
            log.warn("事务回滚，清除暂存事件: eventCount={}", events.size());
            pendingEvents.get().clear();
        }
    }
}

// 在服务中使用
@Service
public class OrderService {
    
    @Autowired
    private TransactionalEventPublisher transactionalEventPublisher;
    
    @Transactional
    public void updateOrderStatus(String orderId, String newStatus) {
        // 更新数据库
        Order order = orderRepository.findById(orderId);
        String oldStatus = order.getStatus();
        order.setStatus(newStatus);
        orderRepository.save(order);
        
        // 在事务中暂存事件
        OrderStatusChangedEvent event = new OrderStatusChangedEvent(
            orderId, order.getOrderNo(), oldStatus, newStatus,
            order.getUserId(), order.getAmount()
        );
        transactionalEventPublisher.publishInTransaction(event);
        
        // 事务提交后，事件会自动发布
    }
}

// 可序列化的事件基类
@Data
@AllArgsConstructor
@NoArgsConstructor
public abstract class SerializableDomainEvent implements Serializable {
    
    private static final long serialVersionUID = 1L;
    
    private String eventId;
    private Long eventTime;
    private String source;
    private String version;
    
    protected SerializableDomainEvent(String source) {
        this.eventId = UUID.randomUUID().toString();
        this.eventTime = System.currentTimeMillis();
        this.source = source;
        this.version = "1.0";
    }
    
    /**
     * 转换为JSON字符串
     */
    public String toJson() {
        try {
            return ObjectMapperHolder.getInstance().writeValueAsString(this);
        } catch (Exception e) {
            throw new EventSerializationException("事件序列化失败", e);
        }
    }
    
    /**
     * 从JSON字符串创建事件对象
     */
    public static <T extends SerializableDomainEvent> T fromJson(String json, Class<T> eventType) {
        try {
            return ObjectMapperHolder.getInstance().readValue(json, eventType);
        } catch (Exception e) {
            throw new EventSerializationException("事件反序列化失败", e);
        }
    }
}

// ObjectMapper持有者（线程安全）
@Component
public class ObjectMapperHolder {
    private static final ObjectMapper INSTANCE;
    
    static {
        INSTANCE = new ObjectMapper();
        INSTANCE.registerModule(new JavaTimeModule());
        INSTANCE.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
        INSTANCE.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    }
    
    public static ObjectMapper getInstance() {
        return INSTANCE;
    }
}

// 事件存储接口
public interface EventStore {
    /**
     * 保存事件
     */
    void saveEvent(DomainEvent event);
    
    /**
     * 获取未处理的事件
     */
    List<DomainEvent> getUnprocessedEvents(String observerName);
    
    /**
     * 标记事件为已处理
     */
    void markEventProcessed(String eventId, String observerName);
    
    /**
     * 获取事件历史
     */
    List<DomainEvent> getEventHistory(String aggregateId, LocalDateTime from, LocalDateTime to);
}

// 基于数据库的事件存储实现
@Repository
@Slf4j
public class DatabaseEventStore implements EventStore {
    
    @Autowired
    private JdbcTemplate jdbcTemplate;
    
    @Override
    public void saveEvent(DomainEvent event) {
        String sql = """
            INSERT INTO domain_event (event_id, event_type, event_data, event_time, source, version)
            VALUES (?, ?, ?, ?, ?, ?)
            """;
        
        try {
            String eventData = event.toJson();
            jdbcTemplate.update(sql, 
                event.getEventId(), 
                event.getClass().getSimpleName(),
                eventData,
                new Timestamp(event.getEventTime()),
                event.getSource(),
                event.getVersion()
            );
            
            log.debug("事件保存成功: eventId={}", event.getEventId());
            
        } catch (Exception e) {
            log.error("事件保存失败: eventId={}", event.getEventId(), e);
            throw new EventStorageException("事件保存失败", e);
        }
    }
    
    @Override
    public List<DomainEvent> getUnprocessedEvents(String observerName) {
        String sql = """
            SELECT de.event_id, de.event_type, de.event_data
            FROM domain_event de
            LEFT JOIN event_processing_log epl ON de.event_id = epl.event_id AND epl.observer_name = ?
            WHERE de.event_id IS NULL
            ORDER BY de.event_time ASC
            LIMIT 100
            """;
        
        try {
            return jdbcTemplate.query(sql, new Object[]{observerName}, (rs, rowNum) -> {
                String eventType = rs.getString("event_type");
                String eventData = rs.getString("event_data");
                
                // 根据事件类型反序列化
                Class<? extends DomainEvent> eventClass = getEventClass(eventType);
                return SerializableDomainEvent.fromJson(eventData, eventClass);
            });
            
        } catch (Exception e) {
            log.error("获取未处理事件失败: observerName={}", observerName, e);
            return Collections.emptyList();
        }
    }
    
    @Override
    public void markEventProcessed(String eventId, String observerName) {
        String sql = """
            INSERT INTO event_processing_log (event_id, observer_name, processed_time)
            VALUES (?, ?, ?)
            ON DUPLICATE KEY UPDATE processed_time = VALUES(processed_time)
            """;
        
        try {
            jdbcTemplate.update(sql, eventId, observerName, new Timestamp(System.currentTimeMillis()));
            log.debug("标记事件处理成功: eventId={}, observerName={}", eventId, observerName);
            
        } catch (Exception e) {
            log.error("标记事件处理失败: eventId={}, observerName={}", eventId, observerName, e);
        }
    }
    
    private Class<? extends DomainEvent> getEventClass(String eventType) {
        // 根据事件类型名称获取对应的Class对象
        // 这里可以通过反射或配置映射来实现
        switch (eventType) {
            case "OrderStatusChangedEvent":
                return OrderStatusChangedEvent.class;
            default:
                throw new IllegalArgumentException("未知的事件类型: " + eventType);
        }
    }
}

// 带持久化的事件发布器
@Component
@Slf4j
public class PersistentEventPublisher extends DefaultDomainEventPublisher {
    
    @Autowired
    private EventStore eventStore;
    
    @Override
    public void publish(DomainEvent event) {
        // 先保存事件
        eventStore.saveEvent(event);
        
        // 再发布事件
        super.publish(event);
    }
    
    /**
     * 重放未处理的事件
     */
    @Scheduled(fixedDelay = 30000) // 每30秒执行一次
    public void replayUnprocessedEvents() {
        List<DomainEventObserver<? extends DomainEvent>> observers = getObservers();
        
        for (DomainEventObserver<? extends DomainEvent> observer : observers) {
            try {
                List<DomainEvent> unprocessedEvents = eventStore.getUnprocessedEvents(observer.getObserverName());
                
                for (DomainEvent event : unprocessedEvents) {
                    try {
                        observer.handle(event);
                        eventStore.markEventProcessed(event.getEventId(), observer.getObserverName());
                        log.debug("事件重放成功: eventId={}, observerName={}", 
                                 event.getEventId(), observer.getObserverName());
                    } catch (Exception e) {
                        log.error("事件重放失败: eventId={}, observerName={}", 
                                 event.getEventId(), observer.getObserverName(), e);
                    }
                }
                
            } catch (Exception e) {
                log.error("重放未处理事件失败: observerName={}", observer.getObserverName(), e);
            }
        }
    }
}

// 事件监控指标
@Component
@Slf4j
public class EventMetrics {
    
    private final MeterRegistry meterRegistry;
    private final Counter eventPublishedCounter;
    private final Counter eventProcessedCounter;
    private final Counter eventFailedCounter;
    private final Timer eventProcessingTimer;
    
    public EventMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.eventPublishedCounter = Counter.builder("event.published")
            .description("事件发布总数")
            .register(meterRegistry);
        this.eventProcessedCounter = Counter.builder("event.processed")
            .description("事件处理成功总数")
            .register(meterRegistry);
        this.eventFailedCounter = Counter.builder("event.failed")
            .description("事件处理失败总数")
            .register(meterRegistry);
        this.eventProcessingTimer = Timer.builder("event.processing.time")
            .description("事件处理时间")
            .register(meterRegistry);
    }
    
    public void recordEventPublished(String eventType) {
        eventPublishedCounter.increment(Tags.of("type", eventType));
    }
    
    public void recordEventProcessed(String eventType, String observerName) {
        eventProcessedCounter.increment(Tags.of("type", eventType, "observer", observerName));
    }
    
    public void recordEventFailed(String eventType, String observerName) {
        eventFailedCounter.increment(Tags.of("type", eventType, "observer", observerName));
    }
    
    public Timer.Sample startTimer() {
        return Timer.start(meterRegistry);
    }
    
    public void recordProcessingTime(Timer.Sample sample, String eventType, String observerName) {
        sample.stop(Timer.builder("event.processing.time")
            .tag("type", eventType)
            .tag("observer", observerName)
            .register(meterRegistry));
    }
}

// 带监控的事件发布器
@Component
@Slf4j
public class MonitoredEventPublisher extends DefaultDomainEventPublisher {
    
    @Autowired
    private EventMetrics eventMetrics;
    
    @Override
    public void publish(DomainEvent event) {
        String eventType = event.getClass().getSimpleName();
        eventMetrics.recordEventPublished(eventType);
        
        List<DomainEventObserver<? extends DomainEvent>> observers = observerMap.get(event.getClass());
        if (observers == null || observers.isEmpty()) {
            return;
        }
        
        for (DomainEventObserver<? extends DomainEvent> observer : observers) {
            Timer.Sample sample = eventMetrics.startTimer();
            try {
                observer.handle(event);
                eventMetrics.recordEventProcessed(eventType, observer.getObserverName());
                eventMetrics.recordProcessingTime(sample, eventType, observer.getObserverName());
            } catch (Exception e) {
                eventMetrics.recordEventFailed(eventType, observer.getObserverName());
                throw e;
            }
        }
    }
}

// 事件统计服务
@Service
@Slf4j
public class EventStatisticsService {
    
    @Autowired
    private EventStore eventStore;
    
    @Autowired
    private MeterRegistry meterRegistry;
    
    /**
     * 获取事件处理统计
     */
    public EventProcessingStatistics getProcessingStatistics(LocalDateTime from, LocalDateTime to) {
        EventProcessingStatistics stats = new EventProcessingStatistics();
        
        // 获取总事件数
        long totalEvents = getTotalEvents(from, to);
        stats.setTotalEvents(totalEvents);
        
        // 获取处理成功数
        long processedEvents = getProcessedEvents(from, to);
        stats.setProcessedEvents(processedEvents);
        
        // 获取处理失败数
        long failedEvents = getFailedEvents(from, to);
        stats.setFailedEvents(failedEvents);
        
        // 计算成功率
        double successRate = totalEvents > 0 ? (double) processedEvents / totalEvents : 0.0;
        stats.setSuccessRate(successRate);
        
        // 获取平均处理时间
        double avgProcessingTime = getAverageProcessingTime(from, to);
        stats.setAverageProcessingTime(avgProcessingTime);
        
        return stats;
    }
    
    /**
     * 获取观察者处理统计
     */
    public List<ObserverStatistics> getObserverStatistics(LocalDateTime from, LocalDateTime to) {
        List<ObserverStatistics> observerStats = new ArrayList<>();
        
        // 获取所有观察者
        List<String> observerNames = getAllObserverNames();
        
        for (String observerName : observerNames) {
            ObserverStatistics stats = new ObserverStatistics();
            stats.setObserverName(observerName);
            
            // 获取该观察者的处理统计
            long processedCount = getObserverProcessedCount(observerName, from, to);
            stats.setProcessedCount(processedCount);
            
            long failedCount = getObserverFailedCount(observerName, from, to);
            stats.setFailedCount(failedCount);
            
            double avgProcessingTime = getObserverAverageProcessingTime(observerName, from, to);
            stats.setAverageProcessingTime(avgProcessingTime);
            
            observerStats.add(stats);
        }
        
        return observerStats;
    }
    
    // 其他统计方法...
}