19消息发布：如何以响应式的编程方式发送消息？

通过上一讲的内容，相信你已经对 Reactive Spring Cloud Stream 的基本架构有了全面的了解。今天这一讲，我将基于 ReactiveSpringCSS 案例系统，带你看看如何使用 Reactive Spring Cloud Stream 来完成响应式消息发布者的构建。

案例集成：消息通信机制与 ReactiveSpringCSS 案例

让我们回到 ReactiveSpringCSS 系统，在案例中分别提取 account-service、order-service 和 customer-service 这三个独立的 Web 服务。显然，它们之间需要进行服务之间的调用和协调，从而完成业务闭环。如果在不久的将来，案例系统中需要引入其他服务才能形成完整的业务流程，那么这个业务闭环背后的交互模式就需要进行相应的调整。

在 ReactiveSpringCSS 案例中，你可以想象一个用户的账户信息变动并不会太频繁。因为 account-service 和 customer-service 分别位于两个服务中，为了降低远程交互的成本，很多时候我们会想到把用户的账户信息集中放在缓存里，并在客户工单生成过程中直接从缓存中获取用户账户。在“16 | Redis 集成：如何实现对 Redis 的响应式数据访问”中，我们已经构建了缓存组件。那么，在这样的设计和实现方式下，试想一旦某个用户账户信息发生变化，我们应该如何正确和高效地应对这一场景呢？

一种思路是系统基于一定的时间间隔定时访问 account-service，来获取用户账户信息并存储到缓存中。但考虑到系统的扩展性，这种服务交互模式并不是一个好的选择，因为用户账户信息更新的时机无法事先预知，而事件驱动架构为我们提供了一种更好的实现方案。

当用户账户信息变更时，account-service 可以发送一个事件，该事件表明了某个用户账户信息已经发生了变化，并将这种变化传递到所有相关的服务，这些服务会根据自身的业务逻辑来消费这一事件。通过这种方式，某个特定服务就可以获取用户信息变更事件，从而正确且高效地更新缓存信息。基于这种设计思路，该场景下的交互示意图如下所示。

用户账户信息更新场景中的事件驱动架构

上图显示，customer-service 会从缓存中获取更新之后的账户信息以便进行后续的业务处理。而为了简单起见，customer-service 同时也扮演了事件消费者的角色，它将获取到的事件中的账户信息存储到缓存中。事件处理架构的优势就在于，当系统中需要添加新的用户账户变更事件处理逻辑来完成整个流程时，我们只需要对该事件添加一个新的消费者即可，而不需要调整 customer-service 中的任何逻辑，这在应对系统扩展性上有很大的优势。

设计 ReactiveSpringCSS 中的消息发布场景

一般而言，事件在命名上通常采用过去时态以表示该事件所代表的动作已经发生。所以，我们把这里的用户信息变更事件命名为 AccountChangedEvent。

接下来我们关注事件发布者 account-service。在 account-service 中需要设计并实现使用 Spring Cloud Stream 发布消息的各个组件，包括 Source、Channel 和 Binder。我们围绕 AccountChangedEvent 事件给出 account-service 内部的整个实现流程，如下图所示。

account-service 消息发布实现流程

在 account-service 中，势必会存在一个对用户账户信息的修改操作，这个修改操作会触发上图中的 AccountChangedEvent 事件，之后该事件将被构建成一个消息并通过 ReactiveAccountChangedSource 进行发送。ReactiveAccountChangedSource 就是一种具体的响应式 Source 实现，然后 ReactiveAccountChangedSource 使用默认的名为“output”的 Channel 进行消息发布。在案例中，我将基于 RabbitMQ 为你演示消息通信过程，所以 Binder 组件封装了这个消息中间件。

构建响应式 Source 组件

站在消息处理的角度讲，消息发布流程并不复杂，主要的实现过程是使用 Reactive Spring Cloud Stream 完成响应式 Source 组件的创建、Binder 组件的配置以及与 account-service 进行集成，让我们一起来看一下。

初始化消息发布环境

无论是消息发布者还是消息消费者，首先都需要引入 spring-cloud-stream 和 spring-cloud-stream-reactive 依赖，如下所示。

1
2
3
4
5
6
7
8
9


<dependency>
     <groupId>org.springframework.cloud</groupId>
     <artifactId>spring-cloud-stream</artifactId>
</dependency>
	 
<dependency>
     <groupId>org.springframework.cloud</groupId>
     <artifactId>spring-cloud-stream-reactive</artifactId>
</dependency>

而在案例中，我们将使用 RabbitMQ 作为消息中间件系统，所以也需要引入 spring-cloud-starter-stream-rabbit 依赖，如下所示。

1
2
3
4


<dependency>
     <groupId>org.springframework.cloud</groupId>
     <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

RabbitMQ 是 AMQP（Advanced Message Queuing Protocol，高级消息队列协议）的典型实现框架。在 RabbitMQ 中，核心概念是交换器（Exchange）。我们可以通过控制交换器与队列之间的路由规则来实现对消息的存储转发、点对点、发布-订阅等消息传递模型。在一个 RabbitMQ 中可能会存在多个队列，交换器如果想要把消息发送到具体某一个队列，就需要通过两者之间的绑定规则来设置路由信息。路由信息的设置是开发人员操控 RabbitMQ 的主要手段，而路由过程的执行依赖于消息头中的路由键（Routing Key）属性。交换器会检查路由键并结合路由算法来决定将消息路由到哪个队列中去。下图就是交换器与队列之间的路由关系图。

RabbitMQ 中交换器与队列的路由关系图

我们通过查看 pom 文件发现，以下组件被添加到了依赖层级关系。显然，Project Reactor、Spring Messaging、AMQP 以及 RabbitMQ 都包含在这个依赖组件列表中。

spring-cloud-stream-reactive 和 spring-cloud-starter-stream-rabbit 中的依赖组件

使用 @EnableBinding 注解

对于消息发布者而言，它在 Spring Cloud Stream 体系中扮演着 Source 的角色，所以我们需要在 account-service 的 Bootstrap 类中标明这个 Spring Boot 应用程序是一个 Source 组件。调整之后的 AccountApplication 类如下所示。

1
2
3
4
5
6
7
8


@SpringCloudApplication
@EnableBinding(Source.class)
public class AccountApplication {
        
    public static void main(String[] args) {
        SpringApplication.run(AccountApplication.class, args);
    }
}

可以看到，我们在原有 AccountApplication 上添加了一个 @EnableBinding(Source.class) 注解，该注解的作用就是告诉 Spring Cloud Stream 这个 Spring Boot 应用程序是一个消息发布者，需要绑定消息中间件，以实现两者之间的连接。@EnableBinding 注解定义比较简单，如下所示。

1
2
3


public @interface EnableBinding {
    Class<?>[] value() default {};
}

我们可以使用一个或者多个接口作为该注解的参数。在上面的代码中，我们使用了 Source 接口，表示与消息中间件绑定的是一个消息发布者。我在下一讲介绍 Sink 组件时，同样也会使用到这个 @EnableBinding 注解。

定义 Event

想要发送事件，首先需要定义事件 AccountChangedEvent，包括事件类型、事件所对应的操作以及事件中包含的业务领域对象。AccountChangedEvent 类的定义如下所示。

1
2
3
4
5
6
7
8
9


public class AccountChangedEvent implements Serializable {
    //事件类型
    private String type;
    //事件所对应的操作（新增、更新和删除）
    private String operation;
    //事件对应的领域模型
    private AccountMessage accountMessage;
    //省略 getter/setter
}

AccountChangedEvent 中的“type”字段代表事件类型，在一个系统中可能存在多种事件，我们通过该字段进行区分。而“operation”字段代表对 Account 的操作类型，当创建 Account 之后，一般的变更操作类型包括更新（Update）和删除（Delete），我们通过该字段区分具体的变更操作。最后“accountMessage”字段用来在不同服务之间传递整个 Account 消息对象。

创建响应式 Source

定义完事件的数据结构之后，接下来我们就需要通过 Source 接口来具体实现消息的发布，我们把消息发布组件命名为 ReactiveAccountChangedSource。在构建这个组件的过程中，我们需要把相关知识点进行串联。在“06 | 流式操作：如何使用 Flux 和 Mono 高效构建响应式数据流”中我们介绍了通过 create() 方法和 FluxSink 对象来创建 Flux，FluxSink 对象能够通过 next() 方法持续产生多个元素，它的使用示例如下所示。

1
2
3
4
5


Flux<Integer> flux = Flux.<Integer> create(fluxSink -> {
    while (true) {
        fluxSink.next(ThreadLocalRandom.current().nextInt());
    }
	}, FluxSink.OverflowStrategy.BUFFER);

上述代码中指定了 FluxSink 的 OverflowStrategy 为 BUFFER，你可以参考“05 | 顶级框架：Spring 为什么选择 Reactor 作为响应式编程框架”的内容来回顾 Reactor 框架中的背压机制。

利用 FluxSink，我们就可以构建出一个用于持续生成 AccountChangedEvent 事件的 Flux 对象，示例代码如下所示。

1
2


private FluxSink<Message<AccountChangedEvent>> eventSink;
private Flux<Message<AccountChangedEvent>> flux = Flux.<Message<AccountChangedEvent>>create(sink -> this.eventSink = sink).publish().autoConnect();

上述代码中，我们首先基于 AccountChangedEvent 事件分别构建了 FluxSink 对象和 Flux 对象，并把两者关联起来；然后使用了 publish() 和 autoConnect() 方法确保一旦 FluxSink 产生数据，Flux 就准备随时发送。

接下来我们构建具体的 AccountChangedEvent 对象，然后通过 FluxSink 的 next() 方法进行消息的发送，代码如下所示。

1
2
3
4
5
6
7
8


AccountChangedEvent originalevent =  new AccountChangedEvent(
    AccountChangedEvent.class.getTypeName(),
    operation,
    accountMessage);
 
Mono<AccountChangedEvent> monoEvent = Mono.just(originalevent);
        
monoEvent.map(event -> eventSink.next(MessageBuilder.withPayload(event).build())).then();

从上述代码中可以注意到，我们还是通过 Spring Messaging 模块所提供的 MessageBuilder 工具类将它转换为消息中间件所能发送的 Message 对象，这是因为整个消息通信机制需要一套统一而抽象的消息定义。在上一讲中，我已经提到在 Spring Cloud Stream 中，这套统一而抽象的消息定义来自 Spring Messaging。

一旦我们具备了一个能够持续生成消息的 Flux 对象，就可以通过上一讲所说的 @StreamEmitter 注解进行消息的发送，示例代码如下所示。

1
2
3
4


@StreamEmitter
public void emit(@Output(Source.OUTPUT) FluxSender output) {
    output.send(this.flux);
}

这里用到了 FluxSender 工具类完成了消息的发送，我们当然也可以使用上一讲介绍的直接返回 Flux 的方法来达到同样的效果。

ReactiveAccountChangedSource 类的完整代码如下所示，其中通过对消息发送过程进行提取，我们对外暴露了 publishReactiveAccountUpdatedEvent() 和 publishReactiveAccountDeletedEvent() 两个方法供业务系统进行使用。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40


@Component
public class ReactiveAccountChangedSource {
 
    private static final Logger logger = LoggerFactory.getLogger(ReactiveAccountChangedSource.class);
  
    private FluxSink<Message<AccountChangedEvent>> eventSink;
    private Flux<Message<AccountChangedEvent>> flux;
    
    public ReactiveAccountChangedSource() {
        this.flux = Flux.<Message<AccountChangedEvent>>create(sink -> this.eventSink = sink).publish().autoConnect();
    }
    
    private Mono<Void> publishAccountChangedEvent(String operation, Account account){
     logger.debug("Sending message for Account Id: {}", account.getId());        
     
     AccountMessage accountMessage = new AccountMessage(account.getId(), account.getAccountCode(), account.getAccountName());
     
     AccountChangedEvent originalevent =  new AccountChangedEvent(
            AccountChangedEvent.class.getTypeName(),
             operation,
             accountMessage);
 
        Mono<AccountChangedEvent> monoEvent = Mono.just(originalevent);
        
        return monoEvent.map(event -> eventSink.next(MessageBuilder.withPayload(event).build())).then();
    }
    
    @StreamEmitter
    public void emit(@Output(Source.OUTPUT) FluxSender output) {
        output.send(this.flux);
    }
    
    public Mono<Void> publishAccountUpdatedEvent(Account account) {
     return publishAccountChangedEvent("UPDATE", account);
    }
    
    public Mono<Void> publishAccountDeletedEvent(Account account) {
     return publishAccountChangedEvent("DELETE", account);
    }
}

配置 Binder

因为我们使用 RabbitMQ 来构建 Spring Cloud Stream 中的 Binder，所以需要在 application.yml 配置文件中添加如下配置项。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


spring:
  cloud:
    stream:
      bindings:
        default:
          content-type: application/json
          binder: rabbitmq
        output:
          destination: account-destination
      binders:
        rabbitmq:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: 127.0.0.1
                port: 5672
                username: guest
                password: guest
	            virtual-host: /  

在以上配置项中，我们创建了一个名为“rabbitmq”的 Binder，并指定了 RabbitMQ 服务器的地址、端口、用户名和密码等信息。同时，我们在设置了 output 的 destination 为 account-destination 之后，会在 RabbitMQ 中创建一个名为“account-destination”的交换器，并把 Spring Cloud Stream 的消息输出通道绑定到该交换器上。

集成服务

最后，我们要做的事情就是在 account-service 中集成消息发布功能，AccountService 会在执行用户账户更新操作的同时，调用 ReactiveAccountChangeSource 完成事件的生成和发送。重构后的 AccountService 类代码如下所示。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34


@Service
public class AccountService {
    
    @Autowired
    private ReactiveAccountRepository accountRepository;
    
    @Autowired
    private ReactiveAccountChangedSource accountChangedSource;
        
    public Mono<Account> getAccountById(String accountId) {
        
        return accountRepository.findById(accountId).log("getAccountById");
    }
    
    public Mono<Account> getAccountByAccountName(String accountName) {
        
        return accountRepository.findAccountByAccountName(accountName).log("getAccountByAccountName");
    }
 
    public Mono<Void> addAccount(Mono<Account> account){
     
     Mono<Account> saveAccount = account.flatMap(accountRepository::save);
     
     return saveAccount.flatMap(accountChangedSource::publishAccountUpdatedEvent);
    }
 
    public Mono<Void> updateAccount(Mono<Account> account){
     
     Mono<Account> saveAccount = account.flatMap(accountRepository::save);
     
     return saveAccount.flatMap(accountChangedSource::publishAccountUpdatedEvent);
	}
	…
}

你可以注意到上述代码中，我们再次使用了 flatMap 操作符完成了消息的发布。

小结与预告

今天这一讲，我们基于用户账户信息更新这一特定的业务场景，介绍了使用 Reactive Spring Cloud Stream 来完成对 ReactiveSpringCSS 系统中消息发布的建模，并提供了针对响应式消息发布者的实现过程。整个消息发布流程还是比较复杂的，涉及多个步骤，你可以结合案例代码进行学习。

这里给你留一道思考题：在使用 Reactive Spring Cloud Stream 发布消息时，如何基于 @StreamEmitter 注解来生成消息流？

下一讲将继续探讨基于 Spring Cloud Stream 的开发过程，我们关注于消息消费者的实现，以及自定义消息通道、消费者分组和消息分区等高级主题的实现方式。

点击链接，获取课程相关代码 ↓↓↓https://github.com/lagoueduCol/ReactiveProgramming-jianxiang.git

-– ### 精选评论

文章目录