Spring Cloud 与响应式微服务

（给ImportNew加星标，提高Java技能）


众所周知，Spring Cloud 服务间的调用方式是使用的 RESTful API，我们平时都是R estTemplate 或 Feign 来调用的，这两种方式其实说到底都是同步的方式。


Spring 支持响应式编程。那么我们能不能在 Spring Cloud 的服务间调用的时候用这种异步非阻塞的方式呢？随着 Spring Cloud Finchley 的发布，这一切均可以实现。


本文我们就用 WebFlux、Spring Data Reactive 从头到脚构建一个响应式的微服务。


准备


为了完成这个示例，我们需要：


  • 一个服务注册中心

  • 两个微服务

  • 一个数据库


服务注册中心：我们要用到服务发现和服务注册，这里用一个单节点的 Eureka Server来做。


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


数据库：因为现在还没几个数据库有实现了反应式数据访问的可用驱动，Spring Data Reactive 目前仅支持 MangoDB、Redis 和 Cassandra，简单起见我们就用 MangoDB。MangoDB 我这里使用 Docker 来创建，一切均用默认配置（主要是懒 ，这样就不用去改 Spring Boot 的配置文件了）


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


实战


服务注册中心


新建一个基本的 Spring Boot 工程，命名为eureka-server。


pom.xml 中依赖如下：


<dependency>    <groupId>org.springframework.cloud</groupId>    <artifactId>spring-cloud-starter-netflix-eureka-server</artifactId></dependency>


配置文件 application.yml 配置如下：


spring:  application:    name: eureka-servereureka:  client:    register-with-eureka: false    fetch-registry: false    service-url:      defaultZone: http://localhost:8000/eureka/server:  port: 8000


在启动类上加上 @EnableEurekaServer 注解：


@EnableEurekaServer@SpringBootApplicationpublic class EurekaServerApplication {
    public static void main(String[] args) {        SpringApplication.run(EurekaServerApplication.class, args);    }}


账户服务


新建一个基本的 Spring Boot 工程，命名为 cloud-account。


如果是使用 Spring Initializr 话，引入 Lombok、Reactive Web、Reactive MongoDB 和 Eureka Discovery 这四个依赖。

最终 pom.xml 中应有以下依赖：


<dependency>    <groupId>org.springframework.boot</groupId>    <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId></dependency><dependency>    <groupId>org.springframework.boot</groupId>    <artifactId>spring-boot-starter-webflux</artifactId></dependency><dependency>    <groupId>org.springframework.cloud</groupId>    <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><dependency>    <groupId>org.projectlombok</groupId>    <artifactId>lombok</artifactId>    <optional>true</optional></dependency><dependency>    <groupId>io.projectreactor</groupId>    <artifactId>reactor-test</artifactId>    <scope>test</scope></dependency>


配置文件 application.yml


spring:  application:    name: cloud-accountserver:  port: 8100eureka:  client:    service-url:      defaultZone: http://localhost:8000/eureka/


创建账户的实体类，其中 @AllArgsConstructor、@NoArgsConstructor 和 @Data 都是 Lombok 提供注解，不了解的可以自行学习，这里不多说了。


@AllArgsConstructor@NoArgsConstructor@Data@Document(collection = "accounts")public class Account {    @Id    private String id;    private String customerId;    private Double amount;}


我们使用 Spring Data Reactive。与非响应式 Spring Data 的 CrudReposity 对应的，响应式的 Spring Data 也提供了相应的 Repository 库：ReactiveCrudReposity，我们也可以使用它的子接口 ReactiveMongoRepository。


public interface AccountMongoReactiveRepository extends ReactiveCrudRepository<Account, String> {    Flux<Account> findByCustomerId(String customerId);}


为账户服务创建对应的 Controller，这里只简单提供一个查询客户的所有账户的接口。为了在后面测试负载均衡，这里加上了调用时间戳的打印。


@RequestMapping("/account")@RestControllerpublic class AccountController {
    @Autowired    private AccountMongoReactiveRepository repository;
    @GetMapping("/customer/{customer}")    public Flux<Account> findByCustomer(@PathVariable(name = "customer") String customer) {        System.out.println("Customer => " + customer + " [ " + LocalDateTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss.SSS")) + " ]");        return repository.findByCustomerId(customer);    }
}


客户服务


新建一个基本的 Spring Boot 工程，命名为cloud-customer，POM 依赖和之前的cloud-account 的一模一样。


配置文件如下，仅是改了服务名和端口号：


spring:  application:    name: cloud-customerserver:  port: 8200eureka:  client:    service-url:      defaultZone: http://localhost:8000/eureka/


创建一个 Customer 的实体类：


@Data@AllArgsConstructor@NoArgsConstructor@Document(collection = "customers")public class Customer {    @Id    private String id;    private String name;    private String mobile;}


数据访问层直接继承 ReactiveCrudRepository，我们便有了基本的 CRUD 能力：


public interface CustomerMongoReactiveRepository extends ReactiveCrudRepository<Customer, String> {}


因为我们只是示例，不做复杂的业务逻辑，所以省略了 Service 层，在 Controller 里边直接将 CRUD 的操作代理给了 Repository。


@RestController@RequestMapping("/customer")public class CustomerController {
    @Autowired private CustomerMongoReactiveRepository repository;    @Autowired private WebClient.Builder webClientBuilder;
    @GetMapping("")    public Flux<Customer> list() {        return repository.findAll();    }
    @GetMapping("/{id}")    public Mono<Customer> get(@PathVariable String id) {        return repository.findById(id);    }
    @PostMapping("")    public Mono<Customer> create(@RequestBody Customer customer) {        return repository.save(customer);    }
    @PutMapping("/{id}")    public Mono<Customer> update(@PathVariable("id") String id, @RequestBody Customer customer) {        customer.setId(id);        return repository.save(customer);    }
    @DeleteMapping("/{id}")    public Mono<Void> delete(@PathVariable String id) {        return repository.deleteById(id);    }
}


到这里，我们的服务注册中心和两个微服务就都好了。但是，这两个微服务之间还是完全独立的，没有相互间的服务调用。现在我们来实现之前说的需求：客户服务与帐户服务可以相互通信，以获取客户的所有帐户，并通过客户服务 API 方法返回。


首先创建一个 Java Config，这里我们不再使用 RestTemplate 来调用服务，而是WebClient。这个配置看起来和注册 RestTemplate 时差不多，但是要注意这里注册的 Bean 是WebClient.Builder。


@Configurationpublic class WebClientConfig {
    @Bean    @LoadBalanced    public WebClient.Builder loadBalancedWebClientBuilder() {        return WebClient.builder();    }
}


除了这种写法，还有一种写法是：


public class MyClass {    @Autowired    private LoadBalancerExchangeFilterFunction lbFunction;
    public Mono<String> doOtherStuff() {        return WebClient.builder().baseUrl("http://cloud-account/account")            .filter(lbFunction)            .build()            .get()            .uri("")            .retrieve()            .bodyToMono(String.class);    }}


下边的是错误的写法，会抛出异常：


@Bean@LoadBalancedpublic WebClient loadBalancedWebClient() {    return WebClient.builder().baseUrl("http://cloud-account/account").build();}



然后在 CustomerController 实现这个端点：


@GetMapping("/{id}/account")public Flux<Account> getAllAccounts(@PathVariable String id) {    return webClientBuilder.baseUrl("http://cloud-account/account/").build()        .get().uri("/customer/" + id)        .retrieve()        .bodyToFlux(Account.class);}

这里需要在 cloud-customer 里创建一个 DTO Account，因为和 cloud-account 里的完全一样，就省略了。


测试


同时启动两个 cloud-account 服务：



然后不断请求 http://localhost:8200/customer/5ae15fa640f1687f200d8941/account 接口，从下图的时间戳可以看出 WebClient 会轮流请求两个服务，达到了负载均衡的效果。



总结


我们从服务提供、服务调用、数据访问三个方面均使用了响应式编程（Reactive Programming），可以说是做到了 Full Reactive Stack Backend。相信你对响应式编程及其在 Web 应用、微服务架构中如何发挥作用有了更多的体会，本文实战是比较基础的，因为切换到响应式思维方式并非易事，希望能够通过上手编写代码体会响应式编程的感觉。


转自：Yibo，

链接：haoyizebo.com/posts/a0cb2c47/


- EOF -

推荐阅读  点击标题可跳转

1、聊聊 Spring data JPA

2、Spring Cloud Stream整合Rabbit之重复投递

3、SpringCloudGateway CORS方案看这篇就够了


看完本文有收获？请转发分享给更多人

关注「ImportNew」，提升Java技能

点赞和在看就是最大的支持❤️


相关推荐

  • Linux 6.2 正式发布：首次原生支持苹果 M1 芯片
  • 为什么研发管理必须要数字化？| 极客时间企业版
  • 被Gartner列为“2023年十大战略技术趋势”，爆火的可观测如何正确落地？
  • 10万字Linux常用命令总结，果断收藏！
  • 大公司为什么禁止在SpringBoot项目中使用@Autowired注解？
  • 阿里高级专家：我对技术架构的理解
  • 聊聊对大模型的一些看法
  • ChatGPT的前世今生：OpenAI的技术「执拗」与「豪赌」
  • ChatGPT发展历程、原理、技术架构详解和产业未来
  • 特征平台：定义、架构、工作流程和开源实现概述
  • 这可能是Spring Boot Starter 讲的最清楚的一次了
  • 换上 HikariCP 连接池，太快了！
  • 一文了解微服务架构的分解设计
  • 美国毒火车事件已致近4.4万只动物死亡；未来不少大学将因人口减少关停；村支书非法占有孤儿生活补贴五年......|酷玩日爆
  • 老丈人能有多瞧不上女婿？ | 每日一冷
  • 征服全球的造纸巨头，为何败走中国？
  • 为什么新加坡人均八万美元了，却并不让人神往
  • 让《三体》从科幻片变成纪录片，怎么做？
  • 不骗你，我的「衣柜」能自动烘干！
  • 官方通报县长涉嫌猥亵他人；“波音再次暂停交付”！；美国俄亥俄州列车脱轨事故首份报告出炉 | 每日大新闻