SpringBoot 开发秘籍 - 事件异步处理

JAVA日知录 一个关注| Java | Spring Boot | Spring Cloud | 干货分享的博客网站 本文由博客端 http://www.javadaily.cn 主动推送

在项目实际开发过程中,我们有很多这样的业务场景:一个事务中处理完一个业务逻辑后需要跟着处理另外一个业务逻辑,伪码大致如下:

@Service
public class ProductServiceImpl {
	...
    public void saveProduct(Product product) {
        productMapper.saveOrder(product);
        notifyService.notify(product);
    }
	...
}

很简单并且很常见的一段业务逻辑:首先将产品先保存数据库,然后发送通知。

某一天你们可能需要把新增的产品存到 Es 中,这时候也需要代码可能变成这样:

@Service
public class ProductServiceImpl {
	...
    public void saveProduct(Product product) {
        productMapper.saveProduct(product);
        esService.saveProduct(product)
        notifyService.notify(product);
    }
	...
}

随着业务需求的变化,代码也需要跟着一遍遍的修改。而且还会存在另外一个问题,如果通知系统挂了,那就不能再新增产品了。

对于上面这种情况非常适合引入消息中间件(消息队列)来对业务进行解耦,但并非所有的业务系统都会引入消息中间件(引入会第三方架构组件会带来很大的运维成本)。

Spring 提供了事件驱动机制可以帮助我们实现这一需求。

Spring 事件驱动

Spring 事件驱动由 3 个部分组成

实现 Spring 事件驱动一般只需要三步:

  1. 自定义需要发布的事件类,需要继承 ApplicationEvent 类
  2. 使用 ApplicationEventPublisher 来发布自定义事件
  3. 使用@EventListener 来监听事件

这里需要特别注意一点,默认情况下事件是同步的。即事件被 publish 后会等待 Listener 的处理。如果发布事件处的业务存在事务,监听器处理也会在相同的事务中。如果需要异步处理事件,可以 onApplicationEvent 方法上加@Aync 支持异步或在有@EventListener 的注解方法上加上@Aync。

image.png

源码实战

public class ProductEvent extends ApplicationEvent {
    public ProductEvent(Product product) {
        super(product);
    }
}
@Service
public class ProductServiceImpl implements IproductService {
	...
    @Autowired
    private ApplicationEventPublisher publisher;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveProduct(Product product) {
		productMapper.saveProduct(product);
        //事件发布
        publisher.publishEvent(product);
    }
    ...
}
@Slf4j
@AllArgsConstructor
public class ProductListener {

	private final NotifyService notifyServcie;

	@Async
	@Order
	@EventListener(ProductEvent.class)
	public void notify(ProductEvent event) {
		Product product = (Product) event.getSource();
		notifyServcie.notify(product, "product");
	}
}
@Slf4j
@EnableSwagger2
@SpringBootApplication
@EnableAsync
public class ApplicationBootstrap {
...
}
@Configuration
public class ExecutorConfig {
    /** 核心线程数 */
    private int corePoolSize = 10;
    /** 最大线程数  */
    private int maxPoolSize = 50;
    /** 队列大小  */
    private int queueCapacity = 10;
    /** 线程最大空闲时间   */
    private int keepAliveSeconds = 150;

    @Bean("customExecutor")
    public Executor myExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setThreadNamePrefix("customExecutor-");
        executor.setKeepAliveSeconds(keepAliveSeconds);

        // rejection-policy:当pool已经达到max size的时候,如何处理新任务
        // CALLER_RUNS:不在新线程中执行任务,而是由调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}
  • Spring

    Spring 是一个开源框架,是于 2003 年兴起的一个轻量级的 Java 开发框架,由 Rod Johnson 在其著作《Expert One-On-One J2EE Development and Design》中阐述的部分理念和原型衍生而来。它是为了解决企业应用开发的复杂性而创建的。框架的主要优势之一就是其分层架构,分层架构允许使用者选择使用哪一个组件,同时为 JavaEE 应用程序开发提供集成的框架。

    810 引用 • 1390 回帖 • 681 关注

赞助商 我要投放

欢迎来到这里!

我们正在构建一个小众社区,大家在这里相互信任,以平等 • 自由 • 奔放的价值观进行分享交流。最终,希望大家能够找到与自己志同道合的伙伴,共同成长。

注册 关于
请输入回帖内容 ...