SPRing Framework 5.0的响应式微服务

xml的主要片段和单个微服务的内容如下。在微服务中，使用Netty来代替了默认的Tomcat服务器。

<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.0.0.BUILD-SNAPSHOT</version></parent><dependencyManagement><dependencies> <dependency> <groupId>org.springframework.boot.experimental</groupId> <artifactId>spring-boot-dependencies-web-reactive</artifactId> <version>0.1.0.BUILD-SNAPSHOT</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies></dependencyManagement><dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId> </dependency> <dependency> <groupId>org.springframework.boot.experimental</groupId> <artifactId>spring-boot-starter-web-reactive</artifactId> <exclusions> <exclusion> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-tomcat</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>io.projectreactor.ipc</groupId> <artifactId>reactor-netty</artifactId> </dependency> <dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> </dependency> <dependency> <groupId>pl.piomin.services</groupId> <artifactId>common</artifactId> <version>${project.version}</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>io.projectreactor.addons</groupId> <artifactId>reactor-test</artifactId> <scope>test</scope> </dependency></dependencies>

有两个微服务：帐户服务和客户服务。每个微服务都有自己的MongoDB数据库，且对外暴露简单的响应式API，用于搜索和保存数据。另外，客户服务与帐户服务可以相互通信，以获取所有的客户帐户，并通过客户服务API方法返回。下面是帐户控制器代码：

@RestControllerpublic class AccountController { @Autowired private AccountRepository repository; @GetMapping(value = "/account/customer/{customer}") public Flux<Account> findByCustomer(@PathVariable("customer") Integer customerId) { return repository.findByCustomerId(customerId) .map(a -> new Account(a.getId(), a.getCustomerId(), a.getNumber(), a.getAmount())); } @GetMapping(value = "/account") public Flux<Account> findAll() { return repository.findAll().map(a -> new Account(a.getId(), a.getCustomerId(), a.getNumber(), a.getAmount())); } @GetMapping(value = "/account/{id}") public Mono<Account> findById(@PathVariable("id") Integer id) { return repository.findById(id) .map(a -> new Account(a.getId(), a.getCustomerId(), a.getNumber(), a.getAmount())); } @PostMapping("/person") public Mono<Account> create(@RequestBody Publisher<Account> accountStream) { return repository .save(Mono.from(accountStream) .map(a -> new pl.piomin.services.account.model.Account(a.getNumber(), a.getCustomerId(), a.getAmount()))) .map(a -> new Account(a.getId(), a.getCustomerId(), a.getNumber(), a.getAmount())); }}

在所有API方法中，还执行common模块从帐户实体（MongoDB @Document注解的）到帐户DTO的映射。下面是帐户存储库类。它使用ReactiveMongoTemplate与Mongo集合进行交互。

@Repositorypublic class AccountRepository { @Autowired private ReactiveMongoTemplate template; public Mono<Account> findById(Integer id) { return template.findById(id, Account.class); } public Flux<Account> findAll() { return template.findAll(Account.class); } public Flux<Account> findByCustomerId(String customerId) { return template.find(query(where("customerId").is(customerId)), Account.class); } public Mono<Account> save(Mono<Account> account) { return template.insert(account); }}

在Spring Boot的main或@Configuration类中，应该为MongoDB声明Spring Bean以及连接设置。

@SpringBootapplicationpublic class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } public @Bean MongoClient mongoClient() { return MongoClients.create("mongodb://192.168.99.100"); } public @Bean ReactiveMongoTemplate reactiveMongoTemplate() { return new ReactiveMongoTemplate(mongoClient(), "account"); }}

使用docker MongoDB容器来处理这个示例。

docker run -d --name mongo -p 27017:27017 mongo

在客户服务中，从帐户服务调用端点的/account/customer/{customer}。在主类中声明为@Bean WebClient。

@Autowiredprivate WebClient webClient;@GetMapping(value = "/customer/accounts/{pesel}")public Mono<Customer> findByPeselWithAccounts(@PathVariable("pesel") String pesel) {return repository.findByPesel(pesel).flatMap(customer -> webClient.get().uri("/account/customer/{customer}", customer.getId()).accept(MediaType.APPLICATION_JSON).exchange().flatMap(response -> response.bodyToFlux(Account.class))).collectList().map(l -> {return new Customer(pesel, l);});}

可以使用Web浏览器或REST客户端来测试GET调用。而用POST，则没有那么简单。下面有两个简单的测试用例，用于添加新客户和获得客户的帐户信息。要测试getCustomerAccounts，需要先在端口2222上运行帐户服务。

@RunWith(SpringRunner.class)@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)public class CustomerTest { private static final Logger logger = Logger.getLogger("CustomerTest"); private WebClient webClient; @LocalServerPort private int port; @Before public void setup() { this.webClient = WebClient.create("http://localhost:" + this.port); } @Test public void getCustomerAccounts() { Customer customer = this.webClient.get().uri("/customer/accounts/234543647565") .accept(MediaType.APPLICATION_JSON).exchange().then(response -> response.bodyToMono(Customer.class)) .block(); logger.info("Customer: " + customer); } @Test public void addCustomer() { Customer customer = new Customer(null, "Adam", "Kowalski", "123456787654"); customer = webClient.post().uri("/customer").accept(MediaType.APPLICATION_JSON) .exchange(BodyInserters.fromObject(customer)).then(response -> response.bodyToMono(Customer.class)) .block(); logger.info("Customer: " + customer); }}

结论

Spring框架开始支持响应式编程非常不错，但现在它还处于早期阶段，也没法与Spring Cloud等项目一起使用。希望在不久的将来，类似服务发现和负载平衡的功能也可用于与Spring响应式微服务相集成。Spring还有一个Spring Cloud Stream项目，支持响应式模型。以后再看吧！