20消息消费：如何选择可用的高级开发技巧？

在上一讲中，我们讨论了 ReactiveSpringCSS 案例中基于 Reactive Spring Cloud Stream 的消息发布场景以及实现方式。今天我将延续上一讲的内容，为你介绍消息消费的应用场景，具体讲解如何在服务中添加消息消费者，以及使用各项消息消费的高级开发技巧。

案例集成：ReactiveSpringCSS 中的消息消费场景

我们继续讨论 ReactiveSpringCSS 案例，根据整个消息交互流程，customer-service 就是 AccountChangedEvent 事件的消费者。根据上一讲讨论的交互流程，customer-service 需要把变更后的用户账户信息更新到 Redis 缓存中。

在 Spring Cloud Stream 中，负责消费消息的是 Sink 组件。因此，我们同样围绕 AccountChangedEvent 事件研究 customer -service 内部的整个实现流程，如下图所示。

customer-service 消息消费实现流程图

在上图中，AccountChangedEvent 事件通过消息中间件发送到 Reactive Spring Cloud Stream 中，Reactive Spring Cloud Stream 通过 Sink 获取消息并交由 ReactiveAccountChangedSink 实现具体的消费逻辑。结合前面提到的消息消费场景下的缓存处理需求，可以想象这个 ReactiveAccountChangedSink 会负责实现缓存相关的处理逻辑。

让我们把消息消费过程与 customer-service 中的业务流程串联起来。我们知道在 customer-service 中存在 ReactiveAccountClient 类，它通过判断缓存中是否存储目标用户账户对象来决定是否需要发起远程调用。我们已经在“16 | Redis 集成：如何实现对 Redis 的响应式数据访问”中构建了 Redis 缓存服务和 ReactiveAccountClient，你可以回顾一下。

下图展示了采用这一设计思想之后的流程。

用户账户更新流程图

在上图中，我们看到 account-service 异步发送的 AccountChangedEvent 事件会被 ReactiveAccountChangedSink 所消费，然后 ReactiveAccountChangedSink 将更新后的用户账户信息存储到缓存以供 customer-service 使用。显然，ReactiveAccountChangedSink 是整个流程的关键。如何实现这个 Sink，就是这一讲的主要内容。

构建响应式 Sink 组件

针对消费者组件，我们采用和消息发布者相同的方式进行实现。首先要说的还是使用 @EnableBinding 注解来初始化消息消费者。

使用 @EnableBinding 注解

与初始化消息发布环境一样，我们同样需要在 customer-service 中引入 spring-cloud-stream、spring-cloud-stream-reactive 以及 spring-cloud-starter-stream-rabbit 这几个 Maven 依赖，并构建 Bootstrap 类。customer-service 中的 Bootstrap 类是 CustomerApplication，其代码如下所示。

1
2
3
4
5
6
7
8

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

显然，对于作为消息消费者的 Bootstrap 类而言，@EnableBinding 注解所绑定的应该是 Sink 接口。那么接下来就到了创建 Sink 这一步。

创建 Sink

在这个过程中，AccountChangedSink 负责处理具体的消息消费逻辑，代码如下所示。

 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

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener; 
...
 
@EnableBinding(Input.class)
public class ReactiveAccountChangedSink {
 
    @Autowired
    AccountRedisRepository accountRedisRepository;
 
    @StreamListener("input")
    public void handleAccountChangedEvent(AccountChangedEvent accountChangedEvent) {
     
     AccountMessage accountMessage = accountChangedEvent.getAccountMessage();
     AccountMapper accountMapper = 
            new AccountMapper(accountMessage.getId(), 
                    accountMessage.getAccountCode(), 
                    accountMessage.getAccountName());
     
        if(accountChangedEvent.getOperation().equals("UPDATE")) {
            accountRedisRepository.updateAccount(accountMapper).subscribe();            
        } else if(accountChangedEvent.getOperation().equals("DELETE")) {
         accountRedisRepository.deleteAccount(accountMapper.getId()).subscribe();
        } else {   
            logger.error("The operations {} is undefined ", accountChangedEvent.getOperation());        }
    }
}

这里使用了 @StreamListener 注解，将该注解添加到某个方法上就可以使之接收处理流中的事件。在上面的例子中，@StreamListener 注解添加在了 handleAccountChangedEvent() 方法上并指向了“input”通道，这意味着所有流经“input”通道的消息都会交由这个 handleAccountChangedEvent() 方法进行处理。

而在 handleAccountChangedEvent() 方法中，我们调用了 Redis 集成那一讲中构建的 AccountRedisRepository 类来完成各种缓存相关的处理。

配置 Binder

对于消息消费者而言，配置 Binder 的方式和消息发布者非常类似。如果使用默认的消息通道，我们只需要把用于发送的“output”通道改为接收的“input”通道就可以了，代码如下所示。

 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
        input:
          destination: account-destination
      binders:
        rabbitmq:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: 127.0.0.1
                port: 5672
                username: guest
                password: guest
                virtual-host: /

到这里，关于如何构建响应式 Sink 组件的实现过程就介绍完毕了。Sink 组件和上一讲介绍的Source 组件构成一组对应关系，所以配置上比较类似。你可以结合上一讲内容做一些类比。

Reactive Spring Cloud Stream 高级开发技巧

在分别介绍完消息发布者和消费者的基本实现过程之后，我们将在此基础上讨论 Reactive Spring Cloud Stream 的高级主题，包括自定义消息通道、使用消费者组以及消息分区。

自定义消息通道

在前面的示例中，无论是消息发布还是消息消费，我们都使用了 Spring Cloud Stream 中默认提供的通道名“output”和“input”。显然，在有些场景下，为了更好地管理系统中存在的所有通道，为通道进行命名是一项最佳实践，这点对于消息消费的场景尤为重要。在接下来的内容中，针对消息消费的场景，我们将不再使用 Sink 组件默认提供的“input”通道，而是尝试通过自定义通道的方式来实现消息消费。

在 Spring Cloud Stream 中，实现一个面向消息消费场景的自定义通道的方法也非常简单，只需要定义一个新的接口，并在该接口中通过 @Input 注解声明一个新的 Channel 即可。例如我们可以定义一个新的 AccountChangedChannel 接口，然后通过 @Input 注解就可以声明一个“accountChangedChannel”通道，代码如下所示。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
 
public interface AccountChangedChannel {
    
    String ACCOUNT_CHANGED = "accountChangedChannel";
    
    @Input(AccountChangedChannel.ACCOUNT_CHANGED)
    SubscribableChannel accountChangedChannel();
}

注意到该通道的类型为 Spring Messaging 中用于消费消息的 SubscribableChannel。同时，我们也注意到这个 AccountChangedChannel 的代码风格与 Spring Cloud Stream 自带的 Sink 接口完全一致。作为回顾，这里我们再看看 Sink 接口的定义，如下所示。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
	 
public interface Sink{
 
    String INPUT = "input";
 
    @Input(Sink.INPUT)
    SubscribableChannel input();
}

当我们完成了自定义的消息通信之后，就可以在 @StreamListener 注解中设置这个通道。以前面介绍的 AccountChangedSink 为例，添加了自定义通道之后的代码重构如下。

1
2
3
4
5
6
7
8

@EnableBinding(AccountChangedChannel.class)
public class ReactiveAccountChangedSink{
 
    @StreamListener(AccountChangedChannel.ACCOUNT_CHANGED)
    public void handleAccountChangedEvent(AccountChangedEvent accountChangedEvent) {
	     …
	}
}

可以看到，这里我们继续使用 @EnableBinding 注解绑定了自定义的 AccountChangedChannel。因为 AccountChangedChannel 中通过 @Input 注解提供了“accountChangedChannel”通道，所以这种用法实际上和 @EnableBinding(Sink.class) 是完全一致的。因此，对于 Binder 的配置而言，我们要做的也只是调整通道的名称就可以了，重构后的 Binder 配置信息如下所示。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

spring:
  cloud:
    stream:
      bindings:
        default:
          content-type: application/json
          binder: rabbitmq
        accountChangedChannel:
          destination: account-destination
      binders:
      ...

对于自定义消息通道而言，我们需要注意的是如何合理规划和设计这些通道的名称。在一个系统中，通常会存在很多自定义通道且这些通道会分散在各个代码工程中。这个时候，按照模块作为前缀来命名通道是一项最佳实践。以案例中的场景为例，如果 account-service 中需要提供多个自定义通道，那么就可以采用类似“account_accountChangedChannel”“account_XXXChannel”这样的命名方式进行统一的管理。

使用消费者分组

在分布式服务架构中，服务多实例部署的场景非常常见。在集群环境下，我们希望服务的不同实例被放置在竞争的消费者关系中，同一服务集群中只有一个实例能够处理给定消息。Spring Cloud Stream 提供的消费者分组可以很方便地实现这一需求，效果图如下所示。

customer-service 消息分组效果示意图

上图中，两个 customer-service 实例构成了一个 customerGroup。在这个 customerGroup 中，AccountChangedEvent 事件只会被一个 customer-service 实例所消费。

要想实现上图所示的消息消费效果，我们唯一要做的事情也是重构 Binder 配置，即在配置 Binder 时指定消费者分组信息即可，如下所示。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

spring:
  cloud:
    stream:
      bindings:
        default:
          content-type: application/json
          binder: rabbitmq
        accountChangedChannel:
          destination: account-destination
	      group: customerGroup
      binders:
      ...

以上基于 RabbitMQ 的配置信息中，我们关注“bindings”段中的通道名称使用了自定义的“accountChangedChannel”，并且在该配置项中设置了“group”为“customerGroup”。

请注意，在分布式环境下，对于 RabbitMQ 等消息中间件而言，使用消费者分组与其说是一个可选性，倒不如说是一个必选项。因为只有采用消费者分组才能确保在集群环境下消息得到正确的消费。所以，在日常应用过程中，通常都建议你使用这项开发技巧。

使用消息分区

最后一项 Spring Cloud Stream 使用上的高级主题是使用消费分区。同样在集群环境下，假设存在两个 customer-service 实例，我们希望用户账户信息中 id 以数字结尾的 AccountChangedEvent 始终由第一个 customer-service 实例进行消费，而 id 以字母结尾的AccountChangedEvent 则始终由第二个 customer-service 实例进行消费。基于类似这样的需求，我们就可以构建消息分区，如下所示。

customer-service 消息分区效果示意图

要想实现上图所示的消息消费效果，我们唯一要做的事情还是重构 account-service 中 Binder 配置，如下所示。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

spring:
  cloud:
    stream:
      bindings:
        default:
          content-type: application/json
          binder: rabbitmq
        output:
          destination: account-destination
          group: customerGroup
          producer:
            partitionKeyExpression: payload.accountMessage.isEndWithDigit()
            partitionCount: 2
      binders:
      ...

先要明确上述配置项针对的是消息发布者 Source 组件，因为我们看到了“producer”配置项。请注意，这里出现了两个新的配置项“partitionKeyExpression”和“partitionCount”，它们就与消息分区有关。其中我们指定了“partitionKeyExpression”为“payload.accountMessage.id.isEndWithDigit()”，这里用到了 Spring 自带的 Spring 表达式语言（SpEL）来对传入的 AccountChangedEvent 进行评估，依据是 AccountMessage 中实现的 isEndWithDigit() 方法，如下所示。

1
2
3
4
5

public boolean isEndWithDigit() {
        
    char last = this.id.charAt(this.id.length() - 1);
    return Character.isDigit(last);
}

我们知道 AccountMessage 中的 id 是一个 UUID 值，所以根据这个 UUID 中最后一位来判断是否是数字。如果返回值为 true，那么只有分区 id 为 1 的 customer-service 能接收到该信息；如果是返回值为 false，则表示只有分区 id 为 2 的 customer-service 能接收到该信息。显然，通过这样的分区策略，分区的数量“partitionCount”应该为 2。

对应的，作为消息消费者的 Sink 组件的配置项如下所示。

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

spring:
  cloud:
    stream:
      bindings:
        default:
          content-type: application/json
          binder: rabbitmq
        accountChangedChannel:
          destination: account-destination
	      group: customerGroup
          consumer:
            partitioned: true
            instanceIndex: 0
            instanceCount: 2
      binders:
      ...

上述配置中同样包含了分区信息，其中 partitioned=true 表示启用消息分区功能、instanceCount = 2 表示消息分区的消费者节点数量为 2 个。而这里的 instanceIndex 参数就是用来设置当前消费者实例的索引号。请注意，instanceIndex 是从 0 开始的，我们在这里就把当前服务实例的索引号设置为 0。显然，在另外一个 customer-service 实例中需要将 instanceIndex 设置为 1。

消息分区是一项高级开发技巧，它也有一定的复杂性。这种复杂性在于引入分区所导致的状态性，即每个消费者实例所消费的消息实际上是根据自身的服务实例索引来确定的。对于分布式环境而言，状态性实际上是需要尽量避免的，因为我们无法确保所有实例都不出现问题。日常开发过程中，建议你谨慎使用消息分区功能。

小结与预告

承接上一讲的内容，今天我们继续讨论了使用 Reactive Spring Cloud Stream 实现消息消费者的方法。同样，我们发现通过合理配置 Binder 组件，这一实现过程也比较简单。此外，Reactive Spring Cloud Stream 中还存在一些高级主题，例如自定义消息通道、消费者组以及消费分区，这一讲同样也介绍了在 ReactiveSpringCSS 案例系统中使用这些高级主题的方法。

最后给你留一道思考题：在 Reactive Spring Cloud Stream 中，如何配置消费者组和消费分区功能？

在介绍完消息驱动架构的设计和实现方法之后，接下来将是整个课程的最后一个主题，即响应式测试。我们将从测试方案开始讲起，看看如何验证响应式编程组件的正确性。

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

-– ### 精选评论 ##### *瑾： > 老师，我看了源码，应该有个优化，在mono下会check是不是callable，如果是，就会subscribeoncallable用publishon的scheduler，应该懂了，谢谢老师 ##### *瑾： > 老师我买了你推荐的spring reactor这本书，在线程这块我自己写代码有个问题没搞明白，SubscribeOn是向上作用，publishon是向下作用，我如果在我的fromCallable的source后面直接加publishon切另一个线程，为什么在source里面生产item的线程不是主线成线程，而是publishon的线程？ ######     讲师回复： >     对的，使用publishon操作符之后，当前线程就切换了publishon的线程了