实习成长之路：下层应用调不到上层方法怎么办？GoogleGuava事件上报来打配合

问题描述

我有一个智能回答自动回复的业务实现，因为项目用到了DDD开发模式，我为这个自动回复做了一个单独的领域，包是在xxx-xxx-xxx-domain包下，也就是领域层，问题是我自动回复领域处理完之后，要代理发送微信消息，发送消息MQ，还有更多的比如：推送WS，商机处理等等，最关键的是这个自动回复有特殊的规则，比如延时发送。发送消息的统一入口sendMsg恰好是在会话service接口中定义的，也就是xxx-xxx-xxx-application层，我这调不到，所以最后采用事件上报，我这里画图给你描述下

也就是说我需要用上层的服务

Google Guava解决

我们把上层需要的参数封装成一个事件对象，让我们的事件处理器发布事件

/**** 关键词自动回复上报Spring事件* @Author huacheng* @Date 2:59 下午 2021/12/20* @Param [xxxxMsgBO, second] second表示延迟秒数* @return void**/private void sendxxxxMsg(List<xxxMsgBO> xxxMsgBO,Integer second){xxxMsgEvent xxxMsgEvent = new xxxMsgEvent();xxxMsgEvent.setSendMsgBO(xxxxMsgBO);xxxMsgEvent.setSecond(second);eventBusHolder.postSync(xxxMsgEvent);}@Slf4j
@Component
public class EventBusHolder implements DisposableBean, ApplicationContextAware {private EventBus syncEventBus;/*** 发布同步事件*/public void postSync(Object event) {try {syncEventBus.post(event);} catch (Exception t) {log.error("unexpected sync post event failed, event:{}", event, t);}}//销毁@Overridepublic void destroy() throws Exception {xxxxPool.shutdown();if (!xxxPool.awaitTermination(30, TimeUnit.SECONDS)) {log.warn("force shutdown eventBusPool");xxxPool.shutdownNow();}}//初始化我们的处理器@Overridepublic void setApplicationContext(ApplicationContext applicationContext) {xxxPool = new ThreadPoolExecutor(8, 8, 0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(10240),new BasicThreadFactory.Builder().namingPattern("event-bus-thread-%d").build(),(r, executor) -> log.warn("event discarded due to overflow"));asyncEventBus = new AsyncEventBus(eventBusPool, (exception, context) -> log.error("async event process failed with subscriber: {} method: {} event: {}",context.getSubscriber().getClass().getSimpleName(), context.getSubscriberMethod().getName(), context.getEvent(), exception));syncEventBus = new EventBus((exception, context) -> log.error("sync event process failed with subscriber: {} method: {} event: {}",context.getSubscriber().getClass().getSimpleName(), context.getSubscriberMethod().getName(), context.getEvent(), exception));applicationContext.getBeansWithAnnotation(EventProcessor.class).values().forEach(bean -> {log.info("register event-bus subscriber: {}", bean.getClass().getSimpleName());asyncEventBus.register(bean);syncEventBus.register(bean);});}
}

怎么配合

/*** 自动回复消息事件处理器* @author 花城* @version 1.0* @Description*/
@EventProcessor   //必须加
@Component
@Slf4j
public class xxxMsgEventHandler {//收到关键词自动回复时间上报动作，放入延迟队列@Subscribepublic void subscribe(AutoReplyMsgEvent event) {DelayQueueManager manager = DelayQueueManager.getInstance();DelayxxxxMsgWorker delayAutoReplySendMsgWorker = new DelayxxxxMsgWorker(event.getxxxMsgBO());manager.put(delayAutoReplySendMsgWorker,event.getSecond(), TimeUnit.SECONDS);}
}

然后我们现在已经到了application层，就可以做我们想做的操作了

@Subscribe注解

Eventbus 翻译过来就是事件总线，用于简化组件和组件，线程和线程之间的消息通信，可以捆成是 Handler + Thread 的替代品。
上面的代码说明了这点

EventBus可以将发布者和订阅者解偶，这就是EventBus最大的方便之处。

我们Subscribe注解标注后就成为了观察者
使用起来较为简单

需要防止并发调用时，可在@Subscribe注解下再加上 @AllowConcurrentEvents
EventBus并未直接实现单例，可以根据自己的业务来随机应对
Subscribe的方法必须是Public的

Guava底层源码实现

这里找了一篇不错的博客
EventBus实现原理