@EventPublisher + @Async 异步事件流详解

news/2024/11/13 23:42:40/

本文主要介绍Spring事件流和@Async异步线程池处理,以及@Async默认线程池可能会导致的问题及解决方法。

事件流

Spring可以使用以观察者模式实现的事件流操作,将业务逻辑解耦,达到职责分离的效果

Spring事件流的详解

发布事件:

public class EmailService implements ApplicationEventPublisherAware {private ApplicationEventPublisher publisher;public void sendEmail(String address, String content) {publisher.publishEvent(new BlackListEvent(this, address, content));// send email...}
}

监听事件:

@EventListener(condition = "#blEvent.content == 'foo'")
public void processBlackListEvent(BlackListEvent blEvent) {// notify appropriate parties via notificationAddress...
}

注意在默认情况下,事件监听器会同步接收事件。这意味着publishEvent()方法将阻塞,直到所有侦听器都已完成对事件的处理为止。

@Async

@Async注解bean的一个方法,就会让它在一个单独的线程中执行。换句话说,调用者不会等待被调用方法的完成

@Async有两个限制:

  1. 它必须仅应用于public方法
  2. 自调用(从同一个类中调用异步方法)将不起作用

原因:该方法需要为public才可以被代理。而自调用是不生效的,因为它绕过了代理,直接调用了底层方法。

异步返回参数

可以通过将实际返回包装在Future中,将@Async应用于具有返回类型的方法

示例详见How To Do @Async in Spring

@Async
public Future<String> asyncMethodWithReturnType() {System.out.println("Execute method asynchronously - " + Thread.currentThread().getName());try {Thread.sleep(5000);return new AsyncResult<String>("hello world !!!!");} catch (InterruptedException e) {//}return null;
}

Spring 还提供了一个实现FutureAsyncResult类。我们可以使用它来跟踪异步方法执行的结果。

现在让我们调用上述方法并使用Future对象检索异步过程的结果。

public void testAsyncAnnotationForMethodsWithReturnType()throws InterruptedException, ExecutionException {System.out.println("Invoking an asynchronous method. " + Thread.currentThread().getName());Future<String> future = asyncAnnotationExample.asyncMethodWithReturnType();while (true) {if (future.isDone()) {System.out.println("Result from asynchronous process - " + future.get());break;}System.out.println("Continue doing something else. ");Thread.sleep(1000);}
}

异步监听器

如果要特定的侦听器异步处理事件,只需重用常规@Async支持:

@EventListener
@Async
public void processBlackListEvent(BlackListEvent event) {// BlackListEvent is processed in a separate thread
}

使用异步事件时,请注意以下限制:

  • 如果事件监听器抛出Exception,它将不会传播给调用者,详见AsyncUncaughtExceptionHandler
  • 此类事件监听器无法发送答复事件。如果您需要发送另一个事件作为处理结果,请注入ApplicationEventPublisher以手动发送事件。

@EventPublisher + @Async 阻塞

@Async注解在使用时,不指定线程池的名称,默认SimpleAsyncTaskExecutor线程池。

默认的线程池配置为核心线程数为8,等待队列为无界队列,即当所有核心线程都在执行任务时,后面的任务会进入队列等待,若逻辑执行速度较慢会导致线程池阻塞,从而出现监听器抛弃和无响应的结果

spring默认线程池配置参数org.springframework.boot.autoconfigure.task.TaskExecutionProperties

/*** Configuration properties for task execution.** @author Stephane Nicoll* @since 2.1.0*/
@ConfigurationProperties("spring.task.execution")
public class TaskExecutionProperties {private final Pool pool = new Pool();/*** Prefix to use for the names of newly created threads.*/private String threadNamePrefix = "task-";public static class Pool {/*** Queue capacity. An unbounded capacity does not increase the pool and therefore* ignores the "max-size" property.*/private int queueCapacity = Integer.MAX_VALUE;/*** Core number of threads.*/private int coreSize = 8;/*** Maximum allowed number of threads. If tasks are filling up the queue, the pool* can expand up to that size to accommodate the load. Ignored if the queue is* unbounded.*/private int maxSize = Integer.MAX_VALUE;/*** Whether core threads are allowed to time out. This enables dynamic growing and* shrinking of the pool.*/private boolean allowCoreThreadTimeout = true;/*** Time limit for which threads may remain idle before being terminated.*/private Duration keepAlive = Duration.ofSeconds(60);//getter/setter}
}

自定义线程池

@Async注解中value参数使用自定义线程池,能让开发工程师更加明确线程池的运行规则,选取适合的线程策略,规避资源耗尽的风险

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略,通常有以下四种策略:

  1. ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常
  2. ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常
  3. ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
  4. ThreadPoolExecutor.CallerRunsPolicy:重试添加当前的任务,自动重复调用 execute() 方法,直到成功
@Configuration
public class ThreadConfig {@Bean("msgThread")public ThreadPoolTaskExecutor getMsgSendTaskExecutor(){ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();taskExecutor.setCorePoolSize(10);taskExecutor.setMaxPoolSize(25);taskExecutor.setQueueCapacity(800);taskExecutor.setAllowCoreThreadTimeOut(false);taskExecutor.setAwaitTerminationSeconds(60);taskExecutor.setThreadNamePrefix("msg-thread-");taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());taskExecutor.initialize();return taskExecutor;}
}

监听事件异步处理

@EventListener(value = MsgEvent.class, condition = "#root.args[0].type == 0")
@Async("msgThread")
public void commonEvent(MsgEvent event) {//logic
}

@Async使用自定义线程池的其他方式


参考资料:

  1. Spring事件流
  2. @Async优化
  3. How To Do @Async in Spring
  4. Spring使用@Async注解

http://www.ppmy.cn/news/1170.html

相关文章

YOLO系列算法改进方法 | 目录一览表

文章目录YOLO系列算法改进方法 | 目录一览表一、注意力机制添加方法二、网络轻量化方法三、优化损失函数四、非极大值抑制五、“TransformerCNN”结构六、特征融合方式改进七、优化锚框生成八、激活函数改进&#x1f4a1;魔改YOLO系列算法&#xff0c;助力涨点&#xff0c;助力…

yolov1算法思想流程简单讲解概述————(究极简单的讲述和理解)

在我想学习算法的时候&#xff0c;我看某些大佬特别喜欢上来就讲论文&#xff0c;给我搞的贼难受&#xff0c;毕竟本人太辣鸡了&#xff0c;上来这么搞看不懂&#xff0c;经过诸多算法的这样折磨。我打算根据自己的亲身经历和学习过程中遇到的问题出一期&#xff0c;先讲算法整…

【Android App】人脸识别中借助摄像头和OpenCV实时检测人脸讲解及实战(附源码和演示 超详细)

需要全部代码请点赞关注收藏后评论区留言私信~~~ 一、借助摄像头实时检测人脸 与Android自带的人脸检测器相比&#xff0c;OpenCV具备更强劲的人脸识别功能&#xff0c;它可以通过摄像头实时检测人脸&#xff0c;实时检测的预览空间是JavaCameraView 常用方法说明如下 setCvC…

屏幕开发学习 -- 迪文串口屏

一 前言 最近学习了一款基于图形化开发的屏幕&#xff0c;在摸索一周后&#xff0c;基本熟悉了这款产品的一个开发过程&#xff0c;今天给大家分享一下迪文串口屏和STM32如何建立通讯&#xff0c;有不足之处&#xff0c;还请见谅&#x1f601; 二 迪文屏介绍 1.选型 我用到的…

必看知识点:Redis 中的原子操作(1)-Redis 中命令的原子性

必看知识点&#xff1a;Redis 中的原子操作(1)-Redis 中命令的原子性 Redis 如何应对并发访问 Redis 中处理并发的方案 业务中有时候我们会用 Redis 处理一些高并发的业务场景&#xff0c;例如&#xff0c;秒杀业务&#xff0c;对于库存的操作。。。 先来分析下&#xff0c…

[附源码]Python计算机毕业设计Django姜太公渔具销售系统

项目运行 环境配置&#xff1a; Pychram社区版 python3.7.7 Mysql5.7 HBuilderXlist pipNavicat11Djangonodejs。 项目技术&#xff1a; django python Vue 等等组成&#xff0c;B/S模式 pychram管理等等。 环境需要 1.运行环境&#xff1a;最好是python3.7.7&#xff0c;…

【Rust日报】2022-12-01 Extism - 使所有软件都可编程扩展

Android 13 上的内存安全语言随着进入 Android 的内存不安全代码的数量减少&#xff0c;内存安全漏洞的数量也随之减少。从 2019 年到 2022 年&#xff0c;安卓系统漏洞的比例从 76% 下降到了 35% 。在 Android 13 中&#xff0c;大约 21% 的新本机代码是由 Rust 编写。AOSP 中…

实现自定义Spring Boot Starter

实现自定义Spring Boot Starter一、原理二、实战1 自定义 Spring Boot Starter1.1 添加maven依赖1.2 属性类AuthorProperties1.3 自动配置类AuthorAutoConfiguration1.4 业务逻辑AuthorServer1.5 spring.factories2 测试自定义的 Spring Boot Starter2.1 新建module或者新建工程…