Spring Framework 5.0的响应式微服务

作者：chszs，未经博主允许不得转载。经许可的转载需注明作者和博客主页：http://blog.csdn.net/chszs

Spring团队已经宣布从5.0版本开始支持响应式编程模型。新的Spring 5.0版本可能会在今年3月发布。幸运的是，包含这些特性的里程碑版本和快照版本（非稳定版）现在可以从Spring存储库获得。另外还有一个新的Spring Web Reactive项目，支持响应式的@Controller注解和一个新的WebClient的客户端响应式。下面，可以进一步看看Spring团队提出的解决方案。

遵循这个文档，见：
http://docs.spring.io/spring-framework/docs/5.0.0.BUILD-SNAPSHOT/spring-framework-reference/html/web-reactive.html

Spring框架在内部使用Reactor实现了对响应式的支持。Reactor是一个Reactive Streams的实现，它进一步扩展了基本的Reactive Streams Publisher以及Flux和Mono composable API，对数据序列0…N和0…1提供了声明式的操作。在服务器端，Spring支持基于注释和函数式编程模型。注释模型使用了@Controller注解和其他同时也支持Spring MVC的注解。对于同步服务，响应式控制器与标准REST控制器是非常相似的，下面说明如何怎样使用注释模型和MongoDB的响应式模块来开发一个简单的响应式微服务。

在例子中，使用了Spring Boot 2.0.0快照版和Spring Web Reactive 0.1.0版。依赖配置pom.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方法返回。下面是帐户控制器代码：

@RestController
public 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集合进行交互。

@Repository
public 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以及连接设置。

@SpringBootApplication
public 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。

@Autowired
private 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项目，支持响应式模型。以后再看吧！