Google Guava和EventBus

// 1. 创建EventBus
@Configuration
public class EventBusConfig {
    @Bean
    public EventBus eventBus() {
        return new EventBus("messageEventBus");
    }
}

// 2. 定义事件
@Data
@AllArgsConstructor
public class MessageEvent {
    private String userId;
    private Message message;
}

// 3. 创建订阅者
@Component
@RequiredArgsConstructor
public class MessageEventListener {
    private final WebSocketHandler webSocketHandler;
    
    @Subscribe  // 标记订阅方法
    public void handleMessageEvent(MessageEvent event) {
        webSocketHandler.sendMessage(event.getUserId(), event.getMessage());
    }
}

// 4. 发布事件
@Service
public class MessageService {
    private final EventBus eventBus;
    
    public void sendMessage(String userId, Message message) {
        // 发布事件
        eventBus.post(new MessageEvent(userId, message));
    }
}

// 创建异步EventBus
@Bean
public AsyncEventBus asyncEventBus() {
    return new AsyncEventBus("asyncMessageEventBus", 
        Executors.newFixedThreadPool(10));
}

// 异步订阅者
@Subscribe
@AllowConcurrentEvents  // 允许并发执行
public void handleAsyncEvent(MessageEvent event) {
    // 异步处理
}

@Subscribe
public void handleDeadEvent(DeadEvent deadEvent) {
    log.warn("未被处理的事件: {}", deadEvent.getEvent());
}

// 基础事件
public class BaseEvent { }

// 特定事件
public class SpecificEvent extends BaseEvent { }

// 订阅者可以处理父类事件
@Subscribe
public void handleBaseEvent(BaseEvent event) {
    // 处理所有继承自BaseEvent的事件
}

public class LoggingEventBus extends EventBus {
    @Override
    public void handleSubscriberException(Throwable e, 
            SubscriberExceptionContext context) {
        log.error("事件处理异常", e);
    }
}

public class EventBus {
    // 事件分发器
    private final Dispatcher dispatcher;
    
    // 订阅者注册表
    private final SubscriberRegistry subscribers;
    
    // 事件发布
    public void post(Object event) {
        // 1. 获取事件分发链
        Iterator<Subscriber> eventSubscribers = 
            subscribers.getSubscribers(event);
            
        // 2. 分发事件
        dispatcher.dispatch(event, eventSubscribers);
    }
}

// 注册订阅者
public void register(Object object) {
    // 1. 查找带有@Subscribe注解的方法
    List<SubscriberMethod> methods = 
        findSubscriberMethods(object);
        
    // 2. 注册到订阅者注册表
    for (SubscriberMethod method : methods) {
        subscribe(object, method);
    }
}

// 按模块定义事件
public class UserEvents {
    @Data @AllArgsConstructor
    public static class UserCreated { }
    
    @Data @AllArgsConstructor
    public static class UserUpdated { }
}

@Subscribe
public void handleEvent(MessageEvent event) {
    try {
        // 处理事件
    } catch (Exception e) {
        log.error("事件处理失败", e);
        // 可以选择重新发布事件或其他处理
    }
}

@Configuration
public class EventBusConfig {
    @Bean
    public AsyncEventBus asyncEventBus() {
        // 使用自定义线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            5, 10, 60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000),
            new ThreadFactoryBuilder()
                .setNameFormat("event-bus-pool-%d")
                .build()
        );
        return new AsyncEventBus(executor);
    }
}

@Aspect
@Component
public class EventBusAspect {
    @Around("execution(* com.example..*.@Subscribe *(..))")
    public Object trackEvent(ProceedingJoinPoint pjp) {
        // 记录事件处理开始
        long start = System.currentTimeMillis();
        
        Object result = pjp.proceed();
        
        // 记录事件处理完成
        log.info("事件处理耗时: {}ms", 
            System.currentTimeMillis() - start);
            
        return result;
    }
}