目录

1. RocketMQ 介绍
2. RocketMQ 基本使用
3. Spring Cloud Stream 介绍
4. 如何使用 Spring Cloud Alibaba RocketMQ Binder
5. Spring Cloud Alibaba RocketMQ Binder 实现
6. MessageSource 支持
7. 配置选项
   1. RocketMQ Binder Properties
   2. RocketMQ Consumer Properties
   3. RocketMQ Provider Properties
8. 官方文档
9. Spring Cloud Stream编写生产者
10. Spring Cloud Stream编写消费者

RocketMQ 介绍

RocketMQ 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。同时，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。

具有以下特点：

• 能够保证严格的消息顺序
• 提供丰富的消息拉取模式
• 高效的订阅者水平扩展能力
• 实时的消息订阅机制
• 亿级消息堆积能力

RocketMQ 基本使用

• 下载 RocketMQ

下载 RocketMQ最新的二进制文件，并解压

解压后的目录结构如下：

apache-rocketmq
├── LICENSE
├── NOTICE
├── README.md
├── benchmark
├── bin
├── conf
└── lib

• 启动 NameServer

nohup sh bin/mqnamesrv &
tail -f ~/logs/rocketmqlogs/namesrv.log

• 启动 Broker

nohup sh bin/mqbroker -n localhost:9876 &
tail -f ~/logs/rocketmqlogs/broker.log

• 发送、接收消息

发送消息：

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

发送成功后显示：SendResult [sendStatus=SEND_OK, msgId= …

接收消息：

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

接收成功后显示：ConsumeMessageThread_%d Receive New Messages: [MessageExt…

• 关闭 Server

sh bin/mqshutdown broker
sh bin/mqshutdown namesrv

Spring Cloud Stream 介绍

Spring Cloud Stream 是一个用于构建基于消息的微服务应用框架。它基于 SpringBoot 来创建具有生产级别的单机 Spring 应用，并且使用 Spring Integration 与 Broker 进行连接。

Spring Cloud Stream 提供了消息中间件配置的统一抽象，推出了 publish-subscribe、consumer groups、partition 这些统一的概念。

Spring Cloud Stream 内部有两个概念：Binder 和 Binding。

• Binder: 跟外部消息中间件集成的组件，用来创建 Binding，各消息中间件都有自己的 Binder 实现。

比如 Kafka 的实现 KafkaMessageChannelBinder，RabbitMQ 的实现 RabbitMessageChannelBinder 以及 RocketMQ 的实现 RocketMQMessageChannelBinder。

• Binding: 包括 Input Binding 和 Output Binding。

Binding 在消息中间件与应用程序提供的 Provider 和 Consumer 之间提供了一个桥梁，实现了开发者只需使用应用程序的 Provider 或 Consumer 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。

Figure 1. Spring Cloud Stream

使用 Spring Cloud Stream 完成一段简单的消息发送和消息接收代码：

MessageChannel messageChannel = new DirectChannel();

// 消息订阅
((SubscribableChannel) messageChannel).subscribe(new MessageHandler() {
    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        System.out.println("receive msg: " + message.getPayload());
    }
});

// 消息发送
messageChannel.send(MessageBuilder.withPayload("simple msg").build());

这段代码所有的消息类都是 spring-messaging 模块里提供的。屏蔽具体消息中间件的底层实现，如果想用更换消息中间件，在配置文件里配置相关消息中间件信息以及修改 binder 依赖即可。

Spring Cloud Stream 底层基于这段代码去做了各种抽象。

如何使用 Spring Cloud Alibaba RocketMQ Binder

如果要在您的项目中引入 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 接收消息。

Note	在使用 RocketMQ Binder 的同时也可以配置 rocketmq.** 用于触发 RocketMQ Spring 相关的 AutoConfiguration

目前 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.

官方文档

https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/2.2.0.RELEASE/spring-cloud-stream.html

Spring Cloud Stream编写生产者

加依赖

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

加注解

@EnableBinding(Source.class)

加配置

spring: 
  cloud:
    stream:
      rocketmq:
        binder:
          name-server: 127.0.0.1:9876
      bindings:
        output:
          # 指定topic
          destination: stream-test-topic

创建消息

@GetMapping("test-stream")
public String testStream(){
    this.source.output().send(
        MessageBuilder.withPayload("消息！！").build()
    );

    return "success";
}

Spring Cloud Stream编写消费者

加依赖

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

加注解

@EnableBinding({Sink.class})

加配置

spring: 
  cloud:
    stream:
      rocketmq:
        binder:
          name-server: 127.0.0.1:9876
      bindings:
        input:
          destination: stream-test-topic
          # rocketmq一定要设置group（随便写） 其他的mq可留空
          group: binding-group

消费消息

package cn.cicoding.usercenter.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Service;

/**
 * @author zhaokejin
 * @description
 * @date 2019/11/15
 */
@Service
@Slf4j
public class TestStreamConsumer {

	@StreamListener(Sink.INPUT)
	public void receive(String messageBody){
		log.info("通过stream收到消息，messageBody = {}", messageBody);
	}


}

启动测试！可以看到控制台收到的消息

2019-12-28 11:54:08.988  INFO 14568 --- [MessageThread_1] c.c.m.consumer.TestStreamConsumer        : 通过stream收到消息，messageBody = RocketMQ消息！！
2019-12-28 11:54:11.814  INFO 14568 --- [MessageThread_1] c.c.m.consumer.TestStreamConsumer        : 通过stream收到消息，messageBody = RocketMQ消息！！
2019-12-28 11:54:12.435  INFO 14568 --- [MessageThread_1] c.c.m.consumer.TestStreamConsumer        : 通过stream收到消息，messageBody = RocketMQ消息！！
2019-12-28 11:54:12.612  INFO 14568 --- [MessageThread_1] c.c.m.consumer.TestStreamConsumer        : 通过stream收到消息，messageBody = RocketMQ消息！！

相关文章

• RocketMQ--消息幂等（去重）通用解决方案
• RocketMQ--消息文件过期原理
• RocketMQ--消息ACK机制及消费进度管理
• RocketMQ--水平扩展及负载均衡详解
• RocketMQ--权限控制