Spring Boot 实现异步事件Event

2040

场景

很多时候当我们完成某些业务后需要给用户推送相关消息提醒。对于这种非核心业务功能我们可以拿出来，创建一个事件去异步执行，从而实现核心业务和子业务的解耦。

实现

定义事件类 Event

创建一个类，继承ApplicationEvent，并重写构造函数。ApplicationEvent是Spring提供的所有应用程序事件扩展类。

public class NoticeEvent extends ApplicationEvent {

    private static final Logger logger = LoggerFactory.getLogger(NoticeEvent.class);

    /**
     * 接受信息
     */
    private String message;

    public NoticeEvent(String message) {
        super(message);
        this.message = message;
        logger.info("add event success! message: {}", message);
    }

    public String getMessage() {
        return message;
    }
}

创建事件监听类 Listener

创建一个用于监听指定事件的类，需要实现ApplicationListener接口，说明它是一个应用程序事件的监听类。注意这里需要加上@Component注解，将其注入Spring容器中。

@Component
public class NoticeListener implements ApplicationListener<NoticeEvent> {

    private static final Logger logger = LoggerFactory.getLogger(NoticeListener.class);

    @Override
    public void onApplicationEvent(NoticeEvent noticeEvent) {
        logger.info("listener get event, sleep 2 second...");
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("event message is : {}", noticeEvent.getMessage());
    }
}

事件发布

事件发布很简单，只需要使用Spring 提供的ApplicationEventPublisher来发布自定义事件

@RestController
@RequestMapping("/event")
public class EventController {

    private static final Logger logger = LoggerFactory.getLogger(EventController.class);

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    @GetMapping("/notice/{message}")
    public void notice(@PathVariable(name = "message") String message) {
        logger.info("begin >>>>>>");
        applicationEventPublisher.publishEvent(new NoticeEvent(message));
        logger.info("end <<<<<<");
    }
}

测试

启动服务，访问接口 http://localhost:8080/event/notice/hello

[nio-8080-exec-1] o.y.n.event.controller.EventController   : begin >>>>>>
[nio-8080-exec-1] org.ylc.note.event.event.NoticeEvent     : add event success! message: hello
[nio-8080-exec-1] org.ylc.note.event.event.NoticeListener  : listener get event, sleep 2 second...
[nio-8080-exec-1] org.ylc.note.event.event.NoticeListener  : event message is : hello
[nio-8080-exec-1] o.y.n.event.controller.EventController   : end <<<<<<

执行成功了，但是通过控制台打印信息发现这里并没有异步执行

异步执行

默认是没有开启异步的，我们需要手动配置开启异步功能，很简单，只需要在配置类上加上@EnableAsync注解就行了，该注解用于声明启用Spring的异步方法执行功能，需要和@Configuration注解一起使用，我们可以直接加在启动类上。然后在监听方法上加上@Async注解，说明当前方法使用异步去执行。

    @Async
    @Override
    public void onApplicationEvent(NoticeEvent noticeEvent) {
        logger.info("listener get event, sleep 2 second...");
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("event message is : {}", noticeEvent.getMessage());
    }

重启服务，重新访问接口：

[nio-8080-exec-1] o.y.n.event.controller.EventController   : begin >>>>>>
[nio-8080-exec-1] org.ylc.note.event.event.NoticeEvent     : add event success! message: hello
[nio-8080-exec-1] o.y.n.event.controller.EventController   : end <<<<<<
[         task-1] org.ylc.note.event.event.NoticeListener  : listener get event, sleep 2 second...
[         task-1] org.ylc.note.event.event.NoticeListener  : event message is : hello

可以发现已经实现了异步功能，主线程为nio-8080-exec-1，监听线程为 task-1。从浏览器反应可以看出，接口直接返回了，并没有等监听线程执行完后才返回。

自定义异步线程池

开启异步后，会使用Spring内部默认的线程池，我们可以自定义这个线程池。创建配置类(加上@Configuration)，实现AsyncConfigurer接口，重写Executor方法。这里我们可以将原先配置在启动类上的@EnableAsync注解放到这个类上。

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    private final Logger log = LoggerFactory.getLogger(this.getClass());

    /**
     * 自定义异步线程池，若不重写会使用默认的线程池
     */
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        // 核心线程数
        taskExecutor.setCorePoolSize(2);
        // 最大线程数
        taskExecutor.setMaxPoolSize(10);
        // 队列大小
        taskExecutor.setQueueCapacity(15);
        // 线程名的前缀
        taskExecutor.setThreadNamePrefix("async-thread-");
        taskExecutor.initialize();
        return taskExecutor;
    }

    /**
     * 捕捉IllegalArgumentException异常
     */
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new MyAsyncExceptionHandler();
    }

    class MyAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
        @Override
        public void handleUncaughtException(Throwable throwable, Method method, Object... objects) {
            log.info("TASK Exception message - " + throwable.getMessage());
            log.info("Method name - " + method.getName());
            for (Object param : objects) {
                log.info("Parameter value - " + param);
            }
        }
    }
}

重启服务，重新访问接口：

[nio-8080-exec-1] o.y.n.event.controller.EventController   : begin >>>>>>
[nio-8080-exec-1] org.ylc.note.event.event.NoticeEvent     : add event success! message: hello
[nio-8080-exec-1] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService
[nio-8080-exec-1] o.y.n.event.controller.EventController   : end <<<<<<
[ async-thread-1] org.ylc.note.event.event.NoticeListener  : listener get event, sleep 2 second...
[ async-thread-1] org.ylc.note.event.event.NoticeListener  : event message is : hello