Spring Boot集成Reactor事件处理框架的简单示例

1. Reactor简介

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

2. Spring Boot集成Reactor的pom.xml

  1 <?xml version="1.0" encoding="UTF-8"?>
  2 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  4     <modelVersion>4.0.0</modelVersion>
  5 
  6     <groupId>org.springframework</groupId>
  7     <artifactId>spring-boot-reactor</artifactId>
  8     <version>0.1.0</version>
  9 
 10     <parent>
 11         <groupId>org.springframework.boot</groupId>
 12         <artifactId>spring-boot-starter-parent</artifactId>
 13         <version>0.5.0.M2</version>
 14     </parent>
 15 
 16     <dependencies>
 17         <dependency>
 18             <groupId>org.springframework.boot</groupId>
 19             <artifactId>spring-boot-starter</artifactId>
 20         </dependency>
 21         <dependency>
 22             <groupId>org.springframework.boot</groupId>
 23             <artifactId>spring-boot-starter-web</artifactId>
 24         </dependency>
 25         <dependency>
 26             <groupId>org.projectreactor</groupId>
 27             <artifactId>reactor-spring</artifactId>
 28         </dependency>
 29         <dependency>
 30             <groupId>com.fasterxml.jackson.core</groupId>
 31             <artifactId>jackson-databind</artifactId>
 32         </dependency>
 33         <dependency>
 34             <groupId>org.springframework</groupId>
 35             <artifactId>spring-web</artifactId>
 36         </dependency>
 37     </dependencies>
 38 
 39     <properties>
 40         <start-class>org.springboot.reactor.example.App</start-class>
 41     </properties>
 42 
 43     <build>
 44         <plugins>
 45             <plugin>
 46                 <artifactId>maven-compiler-plugin</artifactId>
 47             </plugin>
 48             <plugin>
 49                 <groupId>org.springframework.boot</groupId>
 50                 <artifactId>spring-boot-maven-plugin</artifactId>
 51             </plugin>
 52         </plugins>
 53     </build>
 54 
 55     <repositories>
 56         <repository>
 57             <id>spring-snapshots2</id>
 58             <url>http://repo.springsource.org/libs-snapshot</url>
 59             <snapshots>
 60                 <enabled>true</enabled>
 61             </snapshots>
 62         </repository>
 63         <repository>
 64             <id>spring-release</id>
 65             <url>http://repo.springsource.org/libs-release</url>
 66             <snapshots>
 67                 <enabled>false</enabled>
 68             </snapshots>
 69         </repository>
 70     </repositories>
 71     <pluginRepositories>
 72         <pluginRepository>
 73             <id>spring-snapshots</id>
 74             <url>http://repo.springsource.org/libs-snapshot</url>
 75             <snapshots>
 76                 <enabled>true</enabled>
 77             </snapshots>
 78         </pluginRepository>
 79     </pluginRepositories>
 80 </project>

View Code

3. Spring Boot的主程序

  1 package org.springboot.reactor.example;
  2 
  3 import org.springframework.boot.SpringApplication;
  4 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
  5 import org.springframework.context.annotation.ComponentScan;
  6 import org.springframework.context.annotation.Configuration;
  7 
  8 @Configuration
  9 @EnableAutoConfiguration
 10 @ComponentScan
 11 public class App{
 12     public static void main( String[] args ) {
 13     	SpringApplication.run(App.class, args);
 14     }
 15 }

View Code

4. 领域数据User

  1 package org.springboot.reactor.example;
  2 
  3 public class User {
  4 
  5 	String firstName;
  6 
  7 	String lastName;
  8 
  9 	String address;
 10 
 11 	String city;
 12 
 13 }
 14

View Code

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

  1 package org.springboot.reactor.example;
  2 
  3 import org.springframework.context.annotation.Bean;
  4 import org.springframework.context.annotation.Configuration;
  5 
  6 import reactor.core.Environment;
  7 import reactor.core.Reactor;
  8 import reactor.core.spec.Reactors;
  9 
 10 @Configuration
 11 public class Configure {
 12 	@Bean
 13 	Environment env() {
 14 		return new Environment();
 15 	}
 16 
 17 	@Bean
 18     Reactor createReactor(Environment env) {
 19         return Reactors.reactor()
 20                 .env(env)
 21                 .dispatcher(Environment.THREAD_POOL)
 22                 .get();
 23     }
 24 }

View Code

6. Controller层用于通过reactor.notify发送事件

  1 package org.springboot.reactor.example;
  2 
  3 import org.springframework.beans.factory.annotation.Autowired;
  4 import org.springframework.web.bind.annotation.RequestMapping;
  5 import org.springframework.web.bind.annotation.RestController;
  6 
  7 import reactor.core.Reactor;
  8 import reactor.event.Event;
  9 
 10 @RestController
 11 @RequestMapping("/api")
 12 public class ReactorController {
 13 
 14 	@Autowired
 15 	private Reactor reactor;
 16 
 17 	@RequestMapping("/test")
 18 	public void test() throws InterruptedException{
 19 		User user = new User();
 20 		user.firstName = "Chetan";
 21 		user.lastName = "Birajdar";
 22 		user.address = "410 S Hauser";
 23 		user.city = "Los Angeles";
 24 		reactor.notify("eventHandler", Event.wrap(user));
 25 		System.out.println("y0, I sent something for you!!");
 26 	}
 27 }
 28

View Code

7. 事件的监听器，以便于接收发送的事件并处理。需要实现 Consumer<Event<T>> 接口，其中 T 是处理程序接收的数据类型，要根据具体业务设置

  1 package org.springboot.reactor.example;
  2 
  3 import org.springframework.stereotype.Service;
  4 
  5 import reactor.event.Event;
  6 import reactor.function.Consumer;
  7 
  8 @Service
  9 public class AppListener implements Consumer<Event<User>>{
 10 
 11 	public void accept(Event<User> event) {
 12 		System.out.println("Received user object with "
 13 				+ "first name:"
 14 				+ event.getData().firstName
 15 				+ ", last name:"
 16 				+ event.getData().lastName
 17 				+ ", address:"
 18 				+ event.getData().address
 19 				+ ", city:"
 20 				+ event.getData().city
 21 				+ "");
 22 	}
 23 }
 24

View Code

8. ReactorService，实现InitializingBean接口，以便将发送的事件绑定到指定的监听器

  1 package org.springboot.reactor.example;
  2 
  3 import static reactor.event.selector.Selectors.$;
  4 
  5 import org.springframework.beans.factory.InitializingBean;
  6 import org.springframework.beans.factory.annotation.Autowired;
  7 import org.springframework.stereotype.Service;
  8 
  9 import reactor.core.Reactor;
 10 
 11 @Service
 12 public class ReactorService implements InitializingBean{
 13 
 14 	@Autowired
 15 	private Reactor reactor;
 16 
 17 	@Autowired
 18 	private AppListener appListener;
 19 
 20 	@Override
 21 	public void afterPropertiesSet() throws Exception {
 22 		reactor.on($("eventHandler"), appListener);
 23 	}
 24 }

View Code

9. 当启动应用时，调用/api/test接口，向eventHandler的topic发送User事件，监听器即可接收到相关的事件并处理