进阶指南

本章节展示 spring-cloud-statrer-stream-rocketmq 的实现和相关配置。

Spring Cloud Stream RocketMQ Binder 的实现

架构实现

spring-cloud-statrer-stream-rocketmq 去除了对 RocketMQ-Spring 框架的依赖 。 Spring Cloud Stream Binder 核心类 RocketMQMessageChannelBinder 实现了 Spring Cloud Stream 规范，内部会构建 RocketMQInboundChannelAdapter 和 RocketMQProducerMessageHandler。

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

RocketMQInboundChannelAdapter 也会基于 Binding 配置通过 RocketMQConsumerFactory 构造 DefaultMQPushConsumer，其内部会启动 RocketMQ Consumer 接收消息。

NOTE 与 RocketMQ-Spring 框架的兼容需要手动处理。

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

比如 TAGS、KEYS、TRANSACTIONAL_ARGS 等 RocketMQ 消息对应的标签，详情见 com.alibaba.cloud.stream.binder.rocketmq.constant.RocketMQConst

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

或者使用 StreamBridge：

MessageBuilder builder = MessageBuilder.withPayload(msg)
    .setHeader(RocketMQHeaders.TAGS, "binder")
    .setHeader(RocketMQHeaders.KEYS, "my-key");
Message message = builder.build();
streamBridge.send("producer-out-0", message);

NOTE 更多使用请参考样例：com.alibaba.cloud.examples.SenderService。

更多配置项参考

绑定器配置

关于以 spring-cloud-starter-stream-rocketmq-binder 为前缀的配置项如下所示：

消息消费者配置

关于以 spring-cloud-starter-stream-rocketmq-binder-consumer 为前缀的配置项如下所示：

NOTE 更多见 RocketMQConsumerProperties。

关于以 spring-cloud-starter-stream-rocketmq-binder-consumer-push 为前缀的配置项如下所示：

NOTE 其他更多参数见 RocketMQConsumerProperties.Push

关于以 spring-cloud-starter-stream-rocketmq-binder-consumer-pull 为前缀的配置项如下所示：

NOTE 其他更多参数见 RocketMQConsumerProperties.Pull。

消息生产者配置

关于以 spring-cloud-starter-stream-rocketmq-binder-producer 为前缀的配置项如下所示：

NOTE 生产者其他更多参数请见：RocketMQProducerProperties。