大家好我是老三在项目里经常会有一些主线业务之外的其它业务比如下单之后发送通知、监控埋点、记录日志……这些非核心业务如果全部一梭子写下去有两个问题一个是业务耦合一个是串行耗时。下单之后的逻辑所以一般在开发的时候都会把这些操作å抽象成观察者模式也就是发布/订阅模式这里就不讨论观察者模式和发布/订阅模式的不同而且一般会采用多线程的方式来异步执行这些观察者方法。观察者模式一开始我们都是自己去写观察者模式。自己实现观察者模式观察者简图观察者观察者定义接口/** * Author: fighter3 * Description: 观察者接口 * Date: 2022/11/7 11:40 下午 */ public interface OrderObserver { void afterPlaceOrder(PlaceOrderMessage placeOrderMessage); }具体观察者Slf4jpublic class OrderMetricsObserver implements OrderObserver {Overridepublic void afterPlaceOrder(PlaceOrderMessage placeOrderMessage) {log.info([afterPlaceOrder] metrics);}}Slf4jpublic class OrderLogObserver implements OrderObserver{Overridepublic void afterPlaceOrder(PlaceOrderMessage placeOrderMessage) {log.info([afterPlaceOrder] log.);}}Slf4jpublic class OrderNotifyObserver implements OrderObserver{Overridepublic void afterPlaceOrder(PlaceOrderMessage placeOrderMessage) {log.info([afterPlaceOrder] notify.);}}业务通知观察者日志记录观察者监控埋点观察者被观察者消息实体定义Data public class PlaceOrderMessage implements Serializable { /** * 订单号 */ private String orderId; /** * 订单状态 */ private Integer orderStatus; /** * 下单用户ID */ private String userId; //…… }被观察者抽象类public abstract class OrderSubject { //定义一个观察者列表 private ListOrderObserver orderObserverList new ArrayList(); //定义一个线程池,这里参数随便写的 ThreadPoolExecutor threadPoolExecutor new ThreadPoolExecutor(6, 12, 6, TimeUnit.SECONDS, new ArrayBlockingQueue(30)); //增加一个观察者 public void addObserver(OrderObserver o) { this.orderObserverList.add(o); } //删除一个观察者 public void delObserver(OrderObserver o) { this.orderObserverList.remove(o); } //通知所有观察者 public void notifyObservers(PlaceOrderMessage placeOrderMessage) { for (OrderObserver orderObserver : orderObserverList) { //利用多线程异步执行 threadPoolExecutor.execute(() - { orderObserver.afterPlaceOrder(placeOrderMessage); }); } } }这里利用了多线程来异步执行观察者。被观察者实现类/** * Author: fighter3 * Description: 订单实现类-被观察者实现类 * Date: 2022/11/7 11:52 下午 */ Service Slf4j public class OrderServiceImpl extends OrderSubject implements OrderService { /** * 下单 */ Override public PlaceOrderResVO placeOrder(PlaceOrderReqVO reqVO) { PlaceOrderResVO resVO new PlaceOrderResVO(); //添加观察者 this.addObserver(new OrderMetricsObserver()); this.addObserver(new OrderLogObserver()); this.addObserver(new OrderNotifyObserver()); //通知观察者 this.notifyObservers(new PlaceOrderMessage()); log.info([placeOrder] end.); return resVO; } }测试Test DisplayName(下单) void placeOrder() { PlaceOrderReqVO placeOrderReqVO new PlaceOrderReqVO(); orderService.placeOrder(placeOrderReqVO); }测试执行结果2022-11-08 00:11:13.617 INFO 20235 --- [pool-1-thread-1] c.f.obverser.OrderMetricsObserver : [afterPlaceOrder] metrics 2022-11-08 00:11:13.618 INFO 20235 --- [ main] cn.fighter3.obverser.OrderServiceImpl : [placeOrder] end. 2022-11-08 00:11:13.618 INFO 20235 --- [pool-1-thread-3] c.fighter3.obverser.OrderNotifyObserver : [afterPlaceOrder] notify. 2022-11-08 00:11:13.617 INFO 20235 --- [pool-1-thread-2] cn.fighter3.obverser.OrderLogObserver : [afterPlaceOrder] log.可以看到观察者是异步执行的。利用Spring精简可以看到观察者模式写起来还是比较简单的但是既然都用到了Spring来管理Bean的生命周期代码还可以更精简一些。Spring精简观察者模式观察者实现类定义成BeanOrderLogObserverSlf4jServicepublic class OrderLogObserver implements OrderObserver {Overridepublic void afterPlaceOrder(PlaceOrderMessage placeOrderMessage) {log.info([afterPlaceOrder] log.);}}OrderMetricsObserverSlf4j Service public class OrderMetricsObserver implements OrderObserver { Override public void afterPlaceOrder(PlaceOrderMessage placeOrderMessage) { log.info([afterPlaceOrder] metrics); } }OrderNotifyObserverSlf4j Service public class OrderNotifyObserver implements OrderObserver { Override public void afterPlaceOrder(PlaceOrderMessage placeOrderMessage) { log.info([afterPlaceOrder] notify.); } }被观察者自动注入BeanOrderSubjectpublic abstract class OrderSubject {/*** 利用Spring的特性直接注入观察者*/Autowiredprotected ListOrderObserver orderObserverList;//定义一个线程池,这里参数随便写的ThreadPoolExecutor threadPoolExecutor new ThreadPoolExecutor(6, 12, 6, TimeUnit.SECONDS, new ArrayBlockingQueue(30));//通知所有观察者public void notifyObservers(PlaceOrderMessage placeOrderMessage) {for (OrderObserver orderObserver : orderObserverList) {//利用多线程异步执行threadPoolExecutor.execute(() - {orderObserver.afterPlaceOrder(placeOrderMessage);});}}}OrderServiceImplService Slf4j public class OrderServiceImpl extends OrderSubject implements OrderService { /** * 实现类里也要注入一下 */ Autowired private ListOrderObserver orderObserverList; /** * 下单 */ Override public PlaceOrderResVO placeOrder(PlaceOrderReqVO reqVO) { PlaceOrderResVO resVO new PlaceOrderResVO(); //通知观察者 this.notifyObservers(new PlaceOrderMessage()); log.info([placeOrder] end.); return resVO; } }这样一来发现被观察者又简洁了很多但是后来我发现在SpringBoot项目里利用Spring事件驱动驱动模型event模型来实现更加地简练。Spring Event实现发布/订阅模式Spring Event对发布/订阅模式进行了封装使用起来更加简单还是以我们这个场景为例看看怎么来实现吧。自定义事件PlaceOrderEvent继承ApplicationEvent并重写构造函数。ApplicationEvent是Spring提供的所有应用程序事件扩展类。public class PlaceOrderEvent extends ApplicationEvent { public PlaceOrderEvent(PlaceOrderEventMessage source) { super(source); } }PlaceOrderEventMessage事件消息定义了事件的消息体。Data public class PlaceOrderEventMessage implements Serializable { /** * 订单号 */ private String orderId; /** * 订单状态 */ private Integer orderStatus; /** * 下单用户ID */ private String userId; //…… }事件监听者事件监听者有两种实现方式一种是实现ApplicationListener接口另一种是使用EventListener注解。事件监听者实现实现ApplicationListener接口实现ApplicationListener接口,重写onApplicationEvent方法将类定义为Bean这样一个监听者就完成了。OrderLogListenerSlf4j Service public class OrderLogListener implements ApplicationListenerPlaceOrderEvent { Override public void onApplicationEvent(PlaceOrderEvent event) { log.info([afterPlaceOrder] log.); } }OrderMetricsListenerSlf4j Service public class OrderMetricsListener implements ApplicationListenerPlaceOrderEvent { Override public void onApplicationEvent(PlaceOrderEvent event) { log.info([afterPlaceOrder] metrics); } }OrderNotifyListenerSlf4j Service public class OrderNotifyListener implements ApplicationListenerPlaceOrderEvent { Override public void onApplicationEvent(PlaceOrderEvent event) { log.info([afterPlaceOrder] notify.); } }使用EventListener注解使用EventListener注解就更简单了直接在方法上加上EventListener注解就行了。OrderLogListenerSlf4jServicepublic class OrderLogListener {EventListenerpublic void orderLog(PlaceOrderEvent event) {log.info([afterPlaceOrder] log.);}}OrderMetricsListenerSlf4jServicepublic class OrderMetricsListener {EventListenerpublic void metrics(PlaceOrderEvent event) {log.info([afterPlaceOrder] metrics);}}OrderNotifyListenerSlf4jServicepublic class OrderNotifyListener{EventListenerpublic void notify(PlaceOrderEvent event) {log.info([afterPlaceOrder] notify.);}}异步和自定义线程池异步执行异步执行也非常简单使用Spring的异步注解Async就可以了。例如OrderLogListenerSlf4j Service public class OrderLogListener { EventListener Async public void orderLog(PlaceOrderEvent event) { log.info([afterPlaceOrder] log.); } }当然还需要开启异步SpringBoot项目默认是没有开启异步的我们需要手动配置开启异步功能很简单只需要在配置类上加上EnableAsync注解就行了这个注解用于声明启用Spring的异步方法执行功能需要和Configuration注解一起使用也可以直接加在启动类上。SpringBootApplication EnableAsync public class DailyApplication { public static void main(String[] args) { SpringApplication.run(DairlyLearnApplication.class, args); } }自定义线程池使用Async的时候一般都会自定义线程池因为Async的默认线程池为SimpleAsyncTaskExecutor不是真的线程池这个类不重用线程默认每次调用都会创建一个新的线程。自定义线程池有三种方式Async自定义线程池实现接口AsyncConfigurer继承AsyncConfigurerSupport配置由自定义的TaskExecutor替代内置的任务执行器我们来看看三种写法实现接口AsyncConfigurerConfiguration Slf4j public class AsyncConfiguration implements AsyncConfigurer { Bean(fighter3AsyncExecutor) public ThreadPoolTaskExecutor executor() { //Spring封装的一个线程池 ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); //随便写的一些配置 executor.setCorePoolSize(10); executor.setMaxPoolSize(50); executor.setQueueCapacity(30); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.setThreadNamePrefix(fighter3AsyncExecutor-); executor.initialize(); return executor; } Override public Executor getAsyncExecutor() { return executor(); } Override public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { return (ex, method, params) - log.error(String.format([async] task{} error:, method), ex); } }继承AsyncConfigurerSupportConfiguration Slf4j public class SpringAsyncConfigurer extends AsyncConfigurerSupport { Bean public ThreadPoolTaskExecutor asyncExecutor() { ThreadPoolTaskExecutor threadPool new ThreadPoolTaskExecutor(); //随便写的一些配置 threadPool.setCorePoolSize(10); threadPool.setMaxPoolSize(30); threadPool.setWaitForTasksToCompleteOnShutdown(true); threadPool.setAwaitTerminationSeconds(60 * 15); return threadPool; } Override public Executor getAsyncExecutor() { return asyncExecutor(); } Override public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { return (ex, method, params) - log.error(String.format([async] task{} error:, method), ex); } }配置自定义的TaskExecutorSlf4jServicepublic class OrderLogListener {EventListenerAsync(asyncExecutor)public void orderLog(PlaceOrderEvent event) {log.info([afterPlaceOrder] log.);}}配置线程池Configurationpublic class TaskPoolConfig {Bean(name asyncExecutor)public Executor taskExecutor() {ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor();//随便写的一些配置executor.setCorePoolSize(10);executor.setMaxPoolSize(20);executor.setQueueCapacity(200);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix(asyncExecutor-);executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());return executor;}}使用Async注解的时候指定线程池推荐使用这种方式因为在项目里尽量做到线程池隔离不同的任务使用不同的线程池异步和自定义线程池这一部分只是一些扩展稍微占了一些篇幅大家可不要觉得Spring Event用起来很繁琐。发布事件发布事件也非常简单只需要使用Spring 提供的ApplicationEventPublisher来发布自定义事件。OrderServiceImplServiceSlf4jpublic class OrderServiceImpl implements OrderService {Autowiredprivate ApplicationEventPublisher applicationEventPublisher;/*** 下单*/Overridepublic PlaceOrderResVO placeOrder(PlaceOrderReqVO reqVO) {log.info([placeOrder] start.);PlaceOrderResVO resVO new PlaceOrderResVO();//消息PlaceOrderEventMessage eventMessage new PlaceOrderEventMessage();//发布事件applicationEventPublisher.publishEvent(new PlaceOrderEvent(eventMessage));log.info([placeOrder] end.);return resVO;}}在Idea里查看事件的监听者也比较方便点击下面图中的图标就可以查看监听者。查看监听者监听者测试最后我们还是测试一下。Test void placeOrder() { PlaceOrderReqVO placeOrderReqVO new PlaceOrderReqVO(); orderService.placeOrder(placeOrderReqVO); }执行结果2022-11-08 10:05:14.415 INFO 22674 --- [ main] c.f.o.event.event.OrderServiceImpl : [placeOrder] start. 2022-11-08 10:05:14.424 INFO 22674 --- [ main] c.f.o.event.event.OrderServiceImpl : [placeOrder] end. 2022-11-08 10:05:14.434 INFO 22674 --- [sync-executor-3] c.f.o.event.event.OrderNotifyListener : [afterPlaceOrder] notify. 2022-11-08 10:05:14.435 INFO 22674 --- [sync-executor-2] c.f.o.event.event.OrderMetricsListener : [afterPlaceOrder] metrics 2022-11-08 10:05:14.436 INFO 22674 --- [sync-executor-1] c.f.o.event.event.OrderLogListener : [afterPlaceOrder] log.可以看到异步执行而且用到了我们自定义的线程池。小结这篇文章里从最开始自己实现的观察者模式再到利用Spring简化的观察者模式再到使用Spring Event实现发布/订阅模式可以看到Spring Event用起来还是比较简单的。除此之外还有Guava EventBus这样的事件驱动实现大家更习惯使用哪种呢小结这篇文章里从最开始自己实现的观察者模式再到利用Spring简化的观察者模式再到使用Spring Event实现发布/订阅模式可以看到Spring Event用起来还是比较简单的。除此之外还有Guava EventBus这样的事件驱动实现大家更习惯使用哪种呢