Spring Cloud Stream应用RocketMQ

Spring Cloud Stream？

微服务中会经常使用消息中间件，通过消息中间件在服务与服务之间传递消息，例如RabbitMQ、Kafka和RocketMQ，无论使用哪一种消息中间件和服务之间都有一点耦合性，这个耦合性指的是原来使用RabbitMQ，现在要替换为RocketMQ，我们的微服务改动比较大，因为两款消息中间件有一些区别，使用Spring Cloud Stream来整合我们的消息中间件，这样就可以降低微服务和消息中间件的耦合性，做到轻松在不同消息中间件之间切换，然而Spring Cloud Stream官方整合了消息中间件，Spring Cloud Alibaba写了个starter可以支持RocketMQ。
  Spring Cloud Stream是一个构建消息驱动微服务的框架，Spring Cloud Stream解决了开发人员无感知的使用消息中间件的问题，因为Spring Cloud Stream对消息中间件的进一步封装，可以做到代码层面对消息中间件的无感知，甚至于动态的切换中间件(rabbitmq切换为rocketmq或者kafka)，使得微服务开发的高度解耦，服务可以关注更多自己的业务流程。

Spring Cloud Stream 重要概念：

Spring Cloud Stream 内部有几个概念：Binder 、Binding、input、output；
1、Binder：跟外部消息中间件集成的组件，用来创建Binding，各消息中间件都有自己的Binder实现；
例如：Kafka 的实现 KafkaMessageChannelBinder
   RabbitMQ 的实现 RabbitMessageChannelBinder
   RocketMQ 的实现 RocketMQMessageChannelBinder；
2、Binding：包括InputBinding和OutputBinding
  Binding在消息中间件与应用程序提供的Provider和Consumer之间提供了一个桥梁，实现了开发者只需使用应用程序的 Provider 或 Consumer 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触；
3、input：应用程序通过input（相当于消费者consumer）与Spring Cloud Stream 中Binder交互，而Binder负责与消息中间件交互，因此，我们只需关注如何与Binder交互即可，而无需关注与具体消息中间件的交互。
4、output：output（相当于生产者producer）与Spring Cloud Stream中Binder交互；

组成	说明
Binder	Binder是应用与消息中间件之间的封装，目前实现了Kafka和RabbitMQ的Binder，通过Binder可以很方便的连接中间件，可以动态的改变消息类型(对应于Kafka的topic，RabbitMQ的exchange)，这些都可以通过配置文件来实现；
@Input	该注解标识输入通道，通过该输入通道接收消息进入应用程序
@Output	该注解标识输出通道，发布的消息将通过该通道离开应用程序
@StreamListener	监听队列，用于消费者的队列的消息接收
@EnableBinding	将信道channel和exchange、topic绑定在一起

Spring Cloud Stream与RocketMQ应用：

如果要在您的项目中引入 RocketMQ Binder，需要引入如下 maven 依赖：

<dependency>
  <groupId>com.alibaba.cloud</groupId>
  <artifactId>spring-cloud-stream-binder-rocketmq</artifactId>
</dependency>

或者可以使用 Spring Cloud Stream RocketMQ Starter：

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
</dependency>

Spring Cloud Alibaba RocketMQ Binder 实现

RocketMQ Binder 的实现依赖于 RocketMQ-Spring 框架。

RocketMQ-Spring 框架是 RocketMQ 与 Spring Boot 的整合，RocketMQ Spring 主要提供了 3 个特性：

1. 使用 RocketMQTemplate 用来统一发送消息，包括同步、异步发送消息和事务消息
2. @RocketMQTransactionListener 注解用来处理事务消息的监听和回查
3. @RocketMQMessageListener 注解用来消费消息

RocketMQ Binder 的核心类 RocketMQMessageChannelBinder 实现了 Spring Cloud Stream 规范，内部构建会 RocketMQInboundChannelAdapter 和 RocketMQMessageHandler。

RocketMQMessageHandler 会基于 Binding 配置构造 RocketMQTemplate，RocketMQTemplate 内部会把 spring-messaging 模块内 org.springframework.messaging.Message 消息类转换成 RocketMQ 的消息类 org.apache.rocketmq.common.message.Message，然后发送出去。

RocketMQInboundChannelAdapter 也会基于 Binding 配置构造 RocketMQListenerBindingContainer，RocketMQListenerBindingContainer 内部会启动 RocketMQ Consumer 接收消息。

目前 Binder 支持在 Header 中设置相关的 key 来进行 RocketMQ Message 消息的特性设置。

比如 TAGS、DELAY、TRANSACTIONAL_ARG、KEYS、WAIT_STORE_MSG_OK、FLAG 表示 RocketMQ 消息对应的标签，

MessageBuilder builder = MessageBuilder.withPayload(msg)
    .setHeader(RocketMQHeaders.TAGS, "binder")
    .setHeader(RocketMQHeaders.KEYS, "my-key")
    .setHeader("DELAY", "1");
Message message = builder.build();
output().send(message);

MessageSource 支持

目前 RocketMQ 已经支持 MessageSource，可以进行消息的拉取，例子如下：

@SpringBootApplication
@EnableBinding(MQApplication.PolledProcessor.class)
public class MQApplication {

  private final Logger logger =
  	  LoggerFactory.getLogger(MQApplication.class);

  public static void main(String[] args) {
    SpringApplication.run(MQApplication.class, args);
  }

  @Bean
  public ApplicationRunner runner(PollableMessageSource source,
  	    MessageChannel dest) {
    return args -> {
      while (true) {
        boolean result = source.poll(m -> {
          String payload = (String) m.getPayload();
          logger.info("Received: " + payload);
          dest.send(MessageBuilder.withPayload(payload.toUpperCase())
              .copyHeaders(m.getHeaders())
              .build());
        }, new ParameterizedTypeReference<String>() { });
        if (result) {
          logger.info("Processed a message");
        }
        else {
          logger.info("Nothing to do");
        }
        Thread.sleep(5_000);
      }
    };
  }

  public static interface PolledProcessor {

    @Input
    PollableMessageSource source();

    @Output
    MessageChannel dest();

  }

}

配置选项

RocketMQ Binder Properties

• spring.cloud.stream.rocketmq.binder.name-server

  RocketMQ NameServer 地址(老版本使用 namesrv-addr 配置项)。Default: 127.0.0.1:9876.

• spring.cloud.stream.rocketmq.binder.access-key

  阿里云账号 AccessKey。Default: null.

• spring.cloud.stream.rocketmq.binder.secret-key

  阿里云账号 SecretKey。Default: null.

• spring.cloud.stream.rocketmq.binder.enable-msg-trace

  是否为 Producer 和 Consumer 开启消息轨迹功能Default: true.

• spring.cloud.stream.rocketmq.binder.customized-trace-topic

  消息轨迹开启后存储的 topic 名称。Default: RMQ_SYS_TRACE_TOPIC.

RocketMQ Consumer Properties

下面的这些配置是以 spring.cloud.stream.rocketmq.bindings.consumer. 为前缀的 RocketMQ Consumer 相关的配置。

• enable

  是否启用 Consumer。默认值: true.

• tags

  Consumer 基于 TAGS 订阅，多个 tag 以 || 分割。默认值: empty.

• sql

  Consumer 基于 SQL 订阅。默认值: empty.

• broadcasting

  Consumer 是否是广播消费模式。如果想让所有的订阅者都能接收到消息，可以使用广播模式。默认值: false.

• orderly

  Consumer 是否同步消费消息模式。默认值: false.

• delayLevelWhenNextConsume

  异步消费消息模式下消费失败重试策略：-1,不重复，直接放入死信队列0,broker 控制重试策略>0,client 控制重试策略默认值: 0.

• suspendCurrentQueueTimeMillis

  同步消费消息模式下消费失败后再次消费的时间间隔。默认值: 1000.

RocketMQ Provider Properties

下面的这些配置是以 spring.cloud.stream.rocketmq.bindings.producer. 为前缀的 RocketMQ Producer 相关的配置。

• enable

  是否启用 Producer。默认值: true.

• group

  Producer group name。默认值: empty.

• maxMessageSize

  消息发送的最大字节数。默认值: 8249344.

• transactional

  是否发送事务消息。默认值: false.

• sync

  是否使用同步得方式发送消息。默认值: false.

• vipChannelEnabled

  是否在 Vip Channel 上发送消息。默认值: true.

• sendMessageTimeout

  发送消息的超时时间(毫秒)。默认值: 3000.

• compressMessageBodyThreshold

  消息体压缩阀值(当消息体超过 4k 的时候会被压缩)。默认值: 4096.

• retryTimesWhenSendFailed

  在同步发送消息的模式下，消息发送失败的重试次数。默认值: 2.

• retryTimesWhenSendAsyncFailed

  在异步发送消息的模式下，消息发送失败的重试次数。默认值: 2.

• retryNextServer

  消息发送失败的情况下是否重试其它的 broker。默认值: false.

Spring Cloud Stream 默认提供了三个通道接口：Sink、Source 和 Processor，如果当前应用最多只需一个输入通道、一个输出通道，那么直接使用默认通道接口即可，无需再自定义通道接口。

Spring Cloud Stream 默认提供了三个通道接口：Sink、Source 和 Processor，它们源码如下：

（1）三个通道接口说明：

• Sink 和 Source 中分别通过 @Input 和 @Output 注解定义了输入通道和输出通道。
• Processor 通过继承 Source 和 Sink 的方式同时定义了一个输入通道和一个输出通道。

（2）@Input 和 @Output 注解都还有一个 Value 属性，该属性可以用来设置消息通道的名称。没有指定具体的值，将默认使用方法名作为消息通道的名称。

• 因为指定具体的值，Sink 和 Source 的消息通道名称分别为 input 和 output

public` `interface` `Sink {``  ``String INPUT = ``"input"``;` `  ``@Input``(``"input"``)``  ``SubscribableChannel input();``}` `public` `interface` `Source {``  ``String OUTPUT = ``"output"``;` `  ``@Output``(``"output"``)``  ``MessageChannel output();``}` `public` `interface` `Processor ``extends` `Source, Sink {``}