[Spring] SpringBoot 集成 Reactor 事件处理框架

一、 Reactor 框架简介

          Reactor 是 Spring 社区发布的基于事件驱动的异步框架，不仅解耦了程序之间的强调用关系，而且有效提升了系统的多线程并发处理能力。

          基于 SpringBoot 环境使用 Reactor 框架十分方便，下面作简要介绍：

二、 Reactor 基本用法

        2.1  项目基于 SpringBoot + Maven，建立 SpringBoot maven 项目后，增加 Reactor Pom 依赖，主要是 reactor-bus：

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.1.RELEASE</version>
</parent>

<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-bus</artifactId>
</dependency>
</dependencies>

2.2   实例化 EventBus Bean，这里采用内部 Bean 方式实现，其他方式也可以，保证能够供其它类注入即可

import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import reactor.Environment;
import reactor.bus.EventBus;

@Component
public class MyBeans {

@Bean
Environment env() {
return Environment.initializeIfEmpty()
.assignErrorJournal();
}

@Bean
EventBus createEventBus(Environment env) {
return EventBus.create(env, Environment.THREAD_POOL);
}
}

2.3  编写事件处理程序，需要实现 Consumer<Event<T>> 接口，其中 T 是处理程序接收的数据类型，要根据具体业务设置，这里使用 Integer

import org.springframework.stereotype.Service;
import reactor.bus.Event;
import reactor.fn.Consumer;

@Service
class Receiver implements Consumer<Event<Integer>> {

@Override
public void accept(Event<Integer> ev) {
System.out.println("Process number " + ev.getData());
}

}

2.4  关联事件类型与事件处理程序，通过事件标签进行绑定，借助实现 SpringBoot CommandLineRunner 接口，实现启动自动绑定

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import reactor.bus.EventBus;

import static reactor.bus.selector.Selectors.$;

@Component
@Order(value = 1)
public class ReactorLauncher implements CommandLineRunner {

@Autowired
private EventBus eventBus;

@Autowired
private Receiver receiver;

@Override
public void run(String... args) throws Exception {
eventBus.on($("number"), receiver);
}
}

eventBus.on($("number"), receiver); // number类型的事件交给 receiver 处理

2.5  编写事件发送程序，发送程序发送事件时需要指定标签，用以区分不同事件，以便交给不同的处理程序处理

import org.springframework.beans.factory.annotation.Autowired;
import reactor.bus.Event;
import reactor.bus.EventBus;

import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

@Controller
public class Publisher {

@Autowired
EventBus eventBus;

@RequestMapping("reactor/demo")
@ResponseBody
public void publish() throws InterruptedException {

for (int i = 0; i < 10; i++) {
eventBus.notify("number", Event.wrap(i));
}
}

}

eventBus.notify("number", Event.wrap(i)); // 发送number类型事件，将整数 i 作为数据传递给事件处理程序 Receiver

三、 总结

整体来看，使用 Spring Reactor 框架还是比较简单的，只需3步：

1、实现1个事件处理程序

2、绑定事件类型和事件处理程序

3、在需要的地方发送事件