Merge pull request #158 from fangjian0423/1.x

sync code about rocketmq and sentinel for 1.x branch

spring-cloud-alibaba-docs/src/main/asciidoc-zh/rocketmq.adoc

@@ -1 +1,250 @@
-== Spring Cloud Alibaba Rocket Binder
+== Spring Cloud Alibaba RocketMQ Binder
+
+### RocketMQ 介绍
+
+https://rocketmq.apache.org[RocketMQ] 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。同时，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。
+
+具有以下特点：
+
+* 能够保证严格的消息顺序
+
+* 提供丰富的消息拉取模式
+
+* 高效的订阅者水平扩展能力
+
+* 实时的消息订阅机制
+
+* 亿级消息堆积能力
+
+### RocketMQ 基本使用
+
+* 下载 RocketMQ
+
+下载 https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip[RocketMQ最新的二进制文件]，并解压
+
+解压后的目录结构如下：
+
+```
+apache-rocketmq
+├── LICENSE
+├── NOTICE
+├── README.md
+├── benchmark
+├── bin
+├── conf
+└── lib
+```
+
+* 启动 NameServer
+
+```bash
+nohup sh bin/mqnamesrv &
+tail -f ~/logs/rocketmqlogs/namesrv.log
+```
+
+* 启动 Broker
+
+```bash
+nohup sh bin/mqbroker -n localhost:9876 &
+tail -f ~/logs/rocketmqlogs/broker.log
+```
+
+* 发送、接收消息
+
+发送消息：
+
+```bash
+sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
+```
+
+发送成功后显示：`SendResult [sendStatus=SEND_OK, msgId= ...`
+
+接收消息：
+
+```bash
+sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
+```
+
+接收成功后显示：`ConsumeMessageThread_%d Receive New Messages: [MessageExt...`
+
+* 关闭 Server
+
+```bash
+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 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。
+
+.Spring Cloud Stream
+image::https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/images/SCSt-overview.png[]
+
+使用 Spring Cloud Stream 完成一段简单的消息发送和消息接收代码：
+
+```java
+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处理流程
+image::https://cdn.nlark.com/lark/0/2018/png/64647/1543560843558-24525bf4-1d0e-4e10-be5f-bdde7127f6e6.png[]
+
+
+RocketMQ Binder 的核心主要就是这3个类：`RocketMQMessageChannelBinder`，`RocketMQInboundChannelAdapter` 和 `RocketMQMessageHandler`。
+
+`RocketMQMessageChannelBinder` 是个标准的 Binder 实现，其内部构建 `RocketMQInboundChannelAdapter` 和 `RocketMQMessageHandler`。
+
+`RocketMQMessageHandler` 用于 RocketMQ `Producer` 的启动以及消息的发送，其内部会根据 `spring-messaging` 模块内 `org.springframework.messaging.Message` 消息类，去创建 RocketMQ 的消息类 `org.apache.rocketmq.common.message.Message`。
+
+在构造 `org.apache.rocketmq.common.message.Message` 的过程中会根据 `org.springframework.messaging.Message` 的 Header 构造成 `RocketMQMessageHeaderAccessor`。然后再根据 `RocketMQMessageHeaderAccessor` 中的一些属性，比如 tags、keys、flag等属性设置到 RocketMQ 的消息类 `org.apache.rocketmq.common.message.Message` 中。
+
+`RocketMQInboundChannelAdapter` 用于 RocketMQ `Consumer` 的启动以及消息的接收。其内部还支持 https://github.com/spring-projects/spring-retry[spring-retry] 的使用。
+
+在消费消息的时候可以从 Header 中获取 `Acknowledgement` 并进行一些设置。
+
+比如使用 `MessageListenerConcurrently` 进行异步消费的时候，可以设置延迟消费：
+
+```java
+@StreamListener("input")
+public void receive(Message message) {
+    RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message);
+    Acknowledgement acknowledgement = headerAccessor.getAcknowledgement(message);
+    acknowledgement.setConsumeConcurrentlyStatus(ConsumeConcurrentlyStatus.RECONSUME_LATER);
+    acknowledgement.setConsumeConcurrentlyDelayLevel(1);
+}
+```
+
+比如使用 `MessageListenerOrderly` 进行顺序消费的时候，可以设置延迟消费：
+
+```java
+@StreamListener("input")
+public void receive(Message message) {
+    RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message);
+    Acknowledgement acknowledgement = headerAccessor.getAcknowledgement(message);
+    acknowledgement.setConsumeOrderlyStatus(ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT);
+    acknowledgement.setConsumeOrderlySuspendCurrentQueueTimeMill(5000);
+}
+```
+
+Provider端支持的配置：
+
+:frame: topbot
+[width="60%",options="header"]
+|====
+^|配置项 ^|含义 ^| 默认值
+|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.enabled`|是否启用producer|true
+|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.max-message-size`|消息发送的最大字节数|0(大于0才会生效，RocketMQ 默认值为4M = 1024 * 1024 * 4)
+|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.transactional`|是否启用 `TransactionMQProducer` 发送事务消息|false
+|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.executer`|事务消息对应的 `org.apache.rocketmq.client.producer.LocalTransactionExecuter` 接口实现类 全类名。 比如 `org.test.MyExecuter`|
+|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.transaction-check-listener`|事务消息对应的 `org.apache.rocketmq.client.producer.TransactionCheckListener` 接口实现类 全类名。 比如 `org.test.MyTransactionCheckListener`|
+|====
+
+Consumer端支持的配置：
+
+:frame: topbot
+[width="60%",options="header"]
+|====
+^|配置项 ^|含义| 默认值
+|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.enabled`|是否启用consumer|true
+|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.tags`|Consumer订阅只有包括这些tags的topic消息。多个标签之间使用 "\|\|" 分割(不填表示不进行tags的过滤，订阅所有消息)|
+|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.sql`|Consumer订阅满足sql要求的topic消息(如果同时配置了tags内容，sql的优先级更高)|
+|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.broadcasting`|Consumer是否是广播模式|false
+|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.orderly`|顺序消费 or 异步消费|false
+|====
+
+### Endpoint支持
+
+在使用Endpoint特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖，并在配置中允许 Endpoints 的访问。
+
+* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的 endpoint 路径为 `/rocketmq_binder`
+* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的 endpoint 路径为 `/actuator/rocketmq-binder`
+
+Endpoint 会统计消息最后一次发送的数据，消息发送成功或失败的次数，消息消费成功或失败的次数等数据。
+
+```json
+{
+    "runtime": {
+        "lastSend.timestamp": 1542786623915
+    },
+    "metrics": {
+        "scs-rocketmq.consumer.test-topic.totalConsumed": {
+            "count": 11
+        },
+        "scs-rocketmq.consumer.test-topic.totalConsumedFailures": {
+            "count": 0
+        },
+        "scs-rocketmq.producer.test-topic.totalSentFailures": {
+            "count": 0
+        },
+        "scs-rocketmq.consumer.test-topic.consumedPerSecond": {
+            "count": 11,
+            "fifteenMinuteRate": 0.012163847780107841,
+            "fiveMinuteRate": 0.03614605351360527,
+            "meanRate": 0.3493213353657594,
+            "oneMinuteRate": 0.17099243039490175
+        },
+        "scs-rocketmq.producer.test-topic.totalSent": {
+            "count": 5
+        },
+        "scs-rocketmq.producer.test-topic.sentPerSecond": {
+            "count": 5,
+            "fifteenMinuteRate": 0.005540151995103271,
+            "fiveMinuteRate": 0.01652854617838251,
+            "meanRate": 0.10697493212602836,
+            "oneMinuteRate": 0.07995558537067671
+        },
+        "scs-rocketmq.producer.test-topic.sentFailuresPerSecond": {
+            "count": 0,
+            "fifteenMinuteRate": 0.0,
+            "fiveMinuteRate": 0.0,
+            "meanRate": 0.0,
+            "oneMinuteRate": 0.0
+        },
+        "scs-rocketmq.consumer.test-topic.consumedFailuresPerSecond": {
+            "count": 0,
+            "fifteenMinuteRate": 0.0,
+            "fiveMinuteRate": 0.0,
+            "meanRate": 0.0,
+            "oneMinuteRate": 0.0
+        }
+    }
+}
+```
+
+注意：要想查看统计数据需要在pom里加上 https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core[metrics-core依赖]。如若不加，endpoint 将会显示 warning 信息而不会显示统计信息：
+
+```json
+{
+    "warning": "please add metrics-core dependency, we use it for metrics"
+}
+```

spring-cloud-alibaba-docs/src/main/asciidoc-zh/sentinel.adoc

@@ -1,20 +1,20 @@
 == Spring Cloud Alibaba Sentinel
 
-### Sentinel介绍
+### Sentinel 介绍
 
-随着微服务的流行，服务和服务之间的稳定性变得越来越重要。Sentinel 以流量为切入点，从流量控制、熔断降级、系统负载保护等多个维度保护服务的稳定性。
+随着微服务的流行，服务和服务之间的稳定性变得越来越重要。 https://github.com/alibaba/Sentinel[Sentinel] 以流量为切入点，从流量控制、熔断降级、系统负载保护等多个维度保护服务的稳定性。
 
-Sentinel 具有以下特征:
+https://github.com/alibaba/Sentinel[Sentinel] 具有以下特征:
 
 
-* *丰富的应用场景*：Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的核心场景，例如秒杀（即突发流量控制在系统容量可以承受的范围）、消息削峰填谷、实时熔断下游不可用应用等。
+* *丰富的应用场景*： Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的核心场景，例如秒杀（即突发流量控制在系统容量可以承受的范围）、消息削峰填谷、实时熔断下游不可用应用等。
-* *完备的实时监控*：Sentinel 同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒级数据，甚至 500 台以下规模的集群的汇总运行情况。
+* *完备的实时监控*： Sentinel 同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒级数据，甚至 500 台以下规模的集群的汇总运行情况。
-* *广泛的开源生态*：Sentinel 提供开箱即用的与其它开源框架/库的整合模块，例如与 Spring Cloud、Dubbo、gRPC 的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。
+* *广泛的开源生态*： Sentinel 提供开箱即用的与其它开源框架/库的整合模块，例如与 Spring Cloud、Dubbo、gRPC 的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。
-* *完善的 SPI 扩展点*：Sentinel 提供简单易用、完善的 SPI 扩展点。您可以通过实现扩展点，快速的定制逻辑。例如定制规则管理、适配数据源等。
+* *完善的 SPI 扩展点*： Sentinel 提供简单易用、完善的 SPI 扩展点。您可以通过实现扩展点，快速的定制逻辑。例如定制规则管理、适配数据源等。
 
-### 如何使用Sentinel
+### 如何使用 Sentinel
 
-如果要在您的项目中引入Sentinel，使用group ID为 `org.springframework.cloud` 和artifact ID为 `spring-cloud-starter-alibaba-sentinel` 的starter。
+如果要在您的项目中引入 Sentinel，使用 group ID 为 `org.springframework.cloud` 和 artifact ID 为 `spring-cloud-starter-alibaba-sentinel` 的 starter。
 
 ```xml
 <dependency>
@@ -23,7 +23,7 @@ Sentinel 具有以下特征:
 </dependency>
 ```
 
-下面这个例子就是一个最简单的使用Sentinel的例子:
+下面这个例子就是一个最简单的使用 Sentinel 的例子:
 
 ```java
 @SpringBootApplication
@@ -47,9 +47,9 @@ public class TestController {
 }
 ```
 
-@SentinelResource注解用来标识资源是否被限流、降级。上述例子上该注解的属性'hello'表示资源名。
+@SentinelResource 注解用来标识资源是否被限流、降级。上述例子上该注解的属性 'hello' 表示资源名。
 
-@SentinelResource还提供了其它额外的属性如 `blockHandler`，`blockHandlerClass`，`fallback` 用于表示限流或降级的操作，更多内容可以参考 https://github.com/alibaba/Sentinel/wiki/%E6%B3%A8%E8%A7%A3%E6%94%AF%E6%8C%81[Sentinel注解支持]。
+@SentinelResource 还提供了其它额外的属性如 `blockHandler`，`blockHandlerClass`，`fallback` 用于表示限流或降级的操作，更多内容可以参考 https://github.com/alibaba/Sentinel/wiki/%E6%B3%A8%E8%A7%A3%E6%94%AF%E6%8C%81[Sentinel注解支持]。
 
 ##### Sentinel 控制台
 
@@ -74,7 +74,7 @@ image::https://github.com/alibaba/Sentinel/wiki/image/dashboard.png[]
 
 ###### 启动控制台
 
-Sentinel 控制台是一个标准的SpringBoot应用，以SpringBoot的方式运行jar包即可。
+Sentinel 控制台是一个标准的 SpringBoot 应用，以 SpringBoot 的方式运行 jar 包即可。
 
 ```shell
 java -Dserver.port=8080 -Dcsp.sentinel.dashboard.server=localhost:8080 -Dproject.name=sentinel-dashboard -jar sentinel-dashboard.jar
@@ -94,27 +94,73 @@ spring:
         dashboard: localhost:8080
 ----
 
-这里的 `spring.cloud.sentinel.transport.port` 端口配置会在应用对应的机器上启动一个Http Server，该Serve会与Sentinel控制台做交互。比如Sentinel控制台添加了1个限流规则，会把规则数据push给这个Http Server接受，Http Server再将规则注册到Sentinel中。
+这里的 `spring.cloud.sentinel.transport.port` 端口配置会在应用对应的机器上启动一个 Http Server，该 Server 会与 Sentinel 控制台做交互。比如 Sentinel 控制台添加了1个限流规则，会把规则数据 push 给这个 Http Server 接收，Http Server 再将规则注册到 Sentinel 中。
 
-更多Sentinel控制台的使用及问题参考： https://github.com/alibaba/Sentinel/wiki/%E6%8E%A7%E5%88%B6%E5%8F%B0[Sentinel控制台]
+更多 Sentinel 控制台的使用及问题参考： https://github.com/alibaba/Sentinel/wiki/%E6%8E%A7%E5%88%B6%E5%8F%B0[Sentinel控制台]
 
-### Feign支持
+### Feign 支持
 
-DOING
+Sentinel 适配了 https://github.com/OpenFeign/feign[Feign] 组件。如果想使用，除了引入 `sentinel-starter` 的依赖外还需要 3 个步骤：
 
-### RestTemplate支持
+* 配置文件打开 sentinel 对 feign 的支持：`feign.sentinel.enabled=true`
+* 加入 `feign starter` 依赖触发 `sentinel starter` 的配置类生效：
+```xml
+<dependency>
+    <groupId>org.springframework.cloud</groupId>
+    <artifactId>spring-cloud-starter-openfeign</artifactId>
+</dependency>
+```
+* `feign` 内部的 `loadbalance` 功能依赖 Netflix 的 `ribbon` 模块(如果不使用 `loadbalance` 功能可不引入)：
+```xml
+<dependency>
+    <groupId>org.springframework.cloud</groupId>
+    <artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
+</dependency>
+```
 
-Spring Cloud Alibaba Sentinel支持对 `RestTemplate` 的服务调用使用Sentinel进行保护，在构造 `RestTemplate` bean的时候需要加上 `@SentinelProtect` 注解。
+这是一个 `FeignClient` 对应的接口：
+
+```java
+@FeignClient(name = "service-provider", fallback = EchoServiceFallback.class, configuration = FeignConfiguration.class)
+public interface EchoService {
+    @RequestMapping(value = "/echo/{str}", method = RequestMethod.GET)
+    String echo(@PathVariable("str") String str);
+}
+
+class FeignConfiguration {
+    @Bean
+    public EchoServiceFallback echoServiceFallback() {
+        return new EchoServiceFallback();
+    }
+}
+
+class EchoServiceFallback implements EchoService {
+    @Override
+    public String echo(@PathVariable("str") String str) {
+        return "echo fallback";
+    }
+}
+```
+
+NOTE: Feign 对应的接口中的资源名策略定义：httpmethod:protocol://requesturl
+
+`EchoService` 接口中方法 `echo` 对应的资源名为 `GET:http://service-provider/echo/{str}`。
+
+请注意：`@FeignClient` 注解中的所有属性，Sentinel 都做了兼容。
+
+### RestTemplate 支持
+
+Spring Cloud Alibaba Sentinel 支持对 `RestTemplate` 的服务调用使用 Sentinel 进行保护，在构造 `RestTemplate` bean的时候需要加上 `@SentinelRestTemplate` 注解。
 
 ```java
 @Bean
-@SentinelProtect(blockHandler = "handleException", blockHandlerClass = ExceptionUtil.class)
+@SentinelRestTemplate(blockHandler = "handleException", blockHandlerClass = ExceptionUtil.class)
 public RestTemplate restTemplate() {
     return new RestTemplate();
 }
 ```
 
-`@SentinelProtect` 注解的参数跟 `@SentinelResource` 一致，使用方式也一致。
+`@SentinelRestTemplate` 注解的参数跟 `@SentinelResource` 一致，使用方式也一致。
 
 限流的资源规则提供两种粒度：
 
@@ -122,12 +168,13 @@ public RestTemplate restTemplate() {
 
 * `schema://host:port`：协议、主机和端口
 
-NOTE: 以 `https://www.taobao.com/test` 这个url为例。对应的资源名有两种粒度，分别是 `https://www.taobao.com:80` 以及 `https://www.taobao.com:80/test`
+NOTE: 以 `https://www.taobao.com/test` 这个 url 为例。对应的资源名有两种粒度，分别是 `https://www.taobao.com:80` 以及 `https://www.taobao.com:80/test`
-
 
 ### 动态数据源支持
 
-*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之前*，要在Spring Cloud Alibaba Sentinel下使用动态数据源，需要3个步骤：
+#### 在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之前
+
+需要3个步骤才可完成数据源的配置：
 
 * 配置文件中定义数据源信息。比如使用文件：
 
@@ -141,7 +188,7 @@ spring.cloud.sentinel.datasource.converter=flowConverter
 spring.cloud.sentinel.datasource.file=/Users/you/yourrule.json
 ----
 
-* 创建一个Converter类实现 `com.alibaba.csp.sentinel.datasource.Converter` 接口，并且需要在 `ApplicationContext` 中有该类的一个Bean
+* 创建一个 Converter 类实现 `com.alibaba.csp.sentinel.datasource.Converter` 接口，并且需要在 `ApplicationContext` 中有该类的一个 Bean
 
 ```java
 @Component("flowConverter")
@@ -154,7 +201,7 @@ public class JsonFlowRuleListParser implements Converter<String, List<FlowRule>>
 }
 ```
 
-这个Converter的bean name需要跟 `application.properties` 配置文件中的converter配置一致
+这个 Converter 的 bean name 需要跟 `application.properties` 配置文件中的 converter 配置一致
 
 * 在任意一个 Spring Bean 中定义一个被 `@SentinelDataSource` 注解修饰的 `ReadableDataSource` 属性
 
@@ -163,7 +210,7 @@ public class JsonFlowRuleListParser implements Converter<String, List<FlowRule>>
 private ReadableDataSource dataSource;
 ```
 
-`@SentinelDataSource` 注解的value属性表示数据源在 `application.properties` 配置文件中前缀。 该在例子中，前缀为 `spring.cloud.sentinel.datasource` 。
+`@SentinelDataSource` 注解的 value 属性表示数据源在 `application.properties` 配置文件中前缀。 该在例子中，前缀为 `spring.cloud.sentinel.datasource` 。
 
 如果 `ApplicationContext` 中存在超过1个 `ReadableDataSource` bean，那么不会加载这些 `ReadableDataSource` 中的任意一个。 只有在 `ApplicationContext` 存在一个 `ReadableDataSource` 的情况下才会生效。
 
@@ -173,7 +220,9 @@ private ReadableDataSource dataSource;
 [Sentinel Starter] load 3 flow rules
 ```
 
-*在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之后*，要在Spring Cloud Alibaba Sentinel下使用动态数据源，只需要1个步骤：
+#### 在版本 0.2.0.RELEASE 或 0.1.0.RELEASE 之后
+
+只需要1个步骤就可完成数据源的配置：
 
 * 直接在 `application.properties` 配置文件中配置数据源信息即可
 
@@ -196,23 +245,23 @@ spring.cloud.sentinel.datasource.ds4.apollo.default-flow-rule-value = test
 
 ```
 
-这样配置方式参考了Spring Cloud Stream Binder的配置，内部使用了 `TreeMap` 进行存储，comparator为 `String.CASE_INSENSITIVE_ORDER` 。
+这样配置方式参考了 Spring Cloud Stream Binder 的配置，内部使用了 `TreeMap` 进行存储，comparator 为 `String.CASE_INSENSITIVE_ORDER` 。
 
 NOTE: d1, ds2, ds3, ds4 是 `ReadableDataSource` 的名字，可随意编写。后面的 `file` ，`zk` ，`nacos` , `apollo` 就是对应具体的数据源。 它们后面的配置就是这些数据源各自的配置。
 
 每种数据源都有两个共同的配置项： `data-type` 和 `converter-class` 。
 
-`data-type` 配置项表示 `Converter`，Spring Cloud Alibaba Sentinel默认提供两种内置的值，分别是 `json` 和 `xml` (不填默认是json)。 如果不想使用内置的 `json` 或 `xml` 这两种 `Converter`，可以填写 `custom` 表示自定义 `Converter`，然后再配置 `converter-class` 配置项，该配置项需要写类的全路径名。
+`data-type` 配置项表示 `Converter`，Spring Cloud Alibaba Sentinel 默认提供两种内置的值，分别是 `json` 和 `xml` (不填默认是json)。 如果不想使用内置的 `json` 或 `xml` 这两种 `Converter`，可以填写 `custom` 表示自定义 `Converter`，然后再配置 `converter-class` 配置项，该配置项需要写类的全路径名。
 
-这两种内置的 `Converter` 只支持解析 json数组 或 xml数组。内部解析的时候会自动判断每个json对象或xml对象属于哪4种Sentinel规则(`FlowRule`，`DegradeRule`，`SystemRule`，`AuthorityRule`)。
+这两种内置的 `Converter` 只支持解析 json 数组 或 xml 数组。内部解析的时候会自动判断每个 json 对象或xml对象属于哪4种 Sentinel 规则(`FlowRule`，`DegradeRule`，`SystemRule`，`AuthorityRule`)。
 
 比如10个规则数组里解析出5个限流规则和5个降级规则。 这种情况下该数据源不会注册，日志里页会进行警告。
 
 如果10个规则里有9个限流规则，1个解析报错了。这种情况下日志会警告有个规则格式错误，另外9个限流规则会注册上去。
 
-这里json或xml解析用的是 `jackson`。`ObjectMapper` 或 `XmlMapper` 使用默认的配置，遇到不认识的字段会解析报错。 因为不这样做的话限流 json 也会解析成 系统规则(系统规则所有的配置都有默认值)，所以需要这样严格解析。
+这里 json 或 xml 解析用的是 `jackson`。`ObjectMapper` 或 `XmlMapper` 使用默认的配置，遇到不认识的字段会解析报错。 因为不这样做的话限流 json 也会解析成 系统规则(系统规则所有的配置都有默认值)，所以需要这样严格解析。
 
-当然还有一种情况是json对象或xml对象可能会匹配上所有4种规则(比如json对象里只配了 `resource` 字段，那么会匹配上4种规则)，这种情况下日志里会警告，并且过滤掉这个对象。
+当然还有一种情况是 json 对象或 xml 对象可能会匹配上所有4种规则(比如json对象里只配了 `resource` 字段，那么会匹配上4种规则)，这种情况下日志里会警告，并且过滤掉这个对象。
 
 用户使用这种配置的时候只需要填写正确的json或xml就行，有任何不合理的信息都会在日志里打印出来。
 
@@ -223,19 +272,29 @@ NOTE: d1, ds2, ds3, ds4 是 `ReadableDataSource` 的名字，可随意编写。
 [Sentinel Starter] DataSource ds2-sentinel-nacos-datasource load 2 FlowRule
 ```
 
-NOTE: 默认情况下，xml格式是不支持的。需要添加 `jackson-dataformat-xml` 依赖后才会自动生效。
+NOTE: 默认情况下，xml 格式是不支持的。需要添加 `jackson-dataformat-xml` 依赖后才会自动生效。
 
-关于Sentinel动态数据源的实现原理，参考： https://github.com/alibaba/Sentinel/wiki/%E5%8A%A8%E6%80%81%E8%A7%84%E5%88%99%E6%89%A9%E5%B1%95[动态规则扩展]
+关于 Sentinel 动态数据源的实现原理，参考： https://github.com/alibaba/Sentinel/wiki/%E5%8A%A8%E6%80%81%E8%A7%84%E5%88%99%E6%89%A9%E5%B1%95[动态规则扩展]
 
-### Endpoint支持
+### Endpoint 支持
 
-在使用Endpoint特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖，并在配置中允许 Endpoints 的访问。
+在使用 Endpoint 特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖，并在配置中允许 Endpoints 的访问。
 
-* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的endpoint路径为 `/sentinel`
+* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的 endpoint 路径为 `/sentinel`
-* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的endpoint路径为 `/actuator/sentinel`
+* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的 endpoint 路径为 `/actuator/sentinel`
 
 ### More
 
+下表显示当应用的 `ApplicationContext` 中存在对应的Bean的类型时，会进行的一些操作：
+
+:frame: topbot
+[width="60%",options="header"]
+|====
+^|存在Bean的类型 ^|操作 ^|作用
+|`UrlCleaner`|`WebCallbackManager.setUrlCleaner(urlCleaner)`|资源清理(资源（比如将满足 /foo/:id 的 URL 都归到 /foo/* 资源下）)
+|`UrlBlockHandler`|`WebCallbackManager.setUrlBlockHandler(urlBlockHandler)`|自定义限流处理逻辑
+|====
+
 下表显示 Spring Cloud Alibaba Sentinel 的所有配置信息：
 
 :frame: topbot
@@ -244,14 +303,16 @@ NOTE: 默认情况下，xml格式是不支持的。需要添加 `jackson-datafor
 ^|配置项 ^|含义 ^|默认值
 |`spring.cloud.sentinel.enabled`|Sentinel自动化配置是否生效|true
 |`spring.cloud.sentinel.eager`|取消Sentinel控制台懒加载|false
-|`spring.cloud.sentinel.charset`|metric文件字符集|UTF-8
 |`spring.cloud.sentinel.transport.port`|应用与Sentinel控制台交互的端口，应用本地会起一个该端口占用的HttpServer|8721
 |`spring.cloud.sentinel.transport.dashboard`|Sentinel 控制台地址|
 |`spring.cloud.sentinel.transport.heartbeatIntervalMs`|应用与Sentinel控制台的心跳间隔时间|
 |`spring.cloud.sentinel.filter.order`|Servlet Filter的加载顺序。Starter内部会构造这个filter|Integer.MIN_VALUE
 |`spring.cloud.sentinel.filter.spring.url-patterns`|数据类型是数组。表示Servlet Filter的url pattern集合|/*
+|`spring.cloud.sentinel.metric.charset`|metric文件字符集|UTF-8
 |`spring.cloud.sentinel.metric.fileSingleSize`|Sentinel metric 单个文件的大小|
 |`spring.cloud.sentinel.metric.fileTotalCount`|Sentinel metric 总文件数量|
+|`spring.cloud.sentinel.log.dir`|Sentinel 日志文件所在的目录|
+|`spring.cloud.sentinel.log.switch-pid`|Sentinel 日志文件名是否需要带上pid|false
 |`spring.cloud.sentinel.servlet.blockPage`| 自定义的跳转 URL，当请求被限流时会自动跳转至设定好的 URL |
 |`spring.cloud.sentinel.flow.coldFactor`| https://github.com/alibaba/Sentinel/wiki/%E9%99%90%E6%B5%81---%E5%86%B7%E5%90%AF%E5%8A%A8[冷启动因子] |3
 |====

spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/MyTransactionCheckListener.java

@@ -0,0 +1,18 @@
+package org.springframework.cloud.alibaba.cloud.examples;
+
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.client.producer.TransactionCheckListener;
+import org.apache.rocketmq.common.message.MessageExt;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class MyTransactionCheckListener implements TransactionCheckListener {
+
+	@Override
+	public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
+		System.out.println("TransactionCheckListener: " + new String(msg.getBody()));
+		return LocalTransactionState.COMMIT_MESSAGE;
+	}
+
+}

spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/MyTransactionExecuter.java

@@ -0,0 +1,20 @@
+package org.springframework.cloud.alibaba.cloud.examples;
+
+import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.common.message.Message;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class MyTransactionExecuter implements LocalTransactionExecuter {
+	@Override
+	public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
+		if ("1".equals(msg.getUserProperty("test"))) {
+			System.out.println(new String(msg.getBody()) + " rollback");
+			return LocalTransactionState.ROLLBACK_MESSAGE;
+		}
+		System.out.println(new String(msg.getBody()) + " commit");
+		return LocalTransactionState.COMMIT_MESSAGE;
+	}
+}

spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/ReceiveService.java

@@ -30,4 +30,9 @@ public class ReceiveService {
 		System.out.println("input1 receive again: " + receiveMsg);
 	}
 
+	@StreamListener("input4")
+	public void receiveTransactionalMsg(String transactionMsg) {
+		System.out.println("input4 receive transaction msg: " + transactionMsg);
+	}
+
 }

spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQApplication.java

@@ -5,17 +5,19 @@ import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.alibaba.cloud.examples.RocketMQApplication.MySink;
+import org.springframework.cloud.alibaba.cloud.examples.RocketMQApplication.MySource;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.annotation.Input;
-import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.annotation.Output;
 import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 
 /**
  * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
  */
 @SpringBootApplication
-@EnableBinding({ Source.class, MySink.class })
+@EnableBinding({ MySource.class, MySink.class })
 public class RocketMQApplication {
 
 	public interface MySink {
@@ -29,17 +31,33 @@ public class RocketMQApplication {
 		@Input("input3")
 		SubscribableChannel input3();
 
+		@Input("input4")
+		SubscribableChannel input4();
+
+	}
+
+	public interface MySource {
+		@Output("output1")
+		MessageChannel output1();
+
+		@Output("output2")
+		MessageChannel output2();
 	}
 
 	public static void main(String[] args) {
 		SpringApplication.run(RocketMQApplication.class, args);
-    }
+	}
 
 	@Bean
 	public CustomRunner customRunner() {
 		return new CustomRunner();
 	}
 
+	@Bean
+	public CustomRunnerWithTransactional customRunnerWithTransactional() {
+		return new CustomRunnerWithTransactional();
+	}
+
 	public static class CustomRunner implements CommandLineRunner {
 		@Autowired
 		private SenderService senderService;
@@ -62,4 +80,21 @@ public class RocketMQApplication {
 		}
 	}
 
+	public static class CustomRunnerWithTransactional implements CommandLineRunner {
+		@Autowired
+		private SenderService senderService;
+
+		@Override
+		public void run(String... args) throws Exception {
+			// COMMIT_MESSAGE message
+			senderService.sendTransactionalMsg("transactional-msg1", false);
+			// ROLLBACK_MESSAGE message
+			senderService.sendTransactionalMsg("transactional-msg2", true);
+			// ROLLBACK_MESSAGE message
+			senderService.sendTransactionalMsg("transactional-msg3", true);
+			// COMMIT_MESSAGE message
+			senderService.sendTransactionalMsg("transactional-msg4", false);
+		}
+	}
+
 }

spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/SenderService.java

@@ -5,7 +5,7 @@ import java.util.Map;
 
 import org.apache.rocketmq.common.message.MessageConst;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.cloud.alibaba.cloud.examples.RocketMQApplication.MySource;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.MessageBuilder;
@@ -19,17 +19,17 @@ import org.springframework.util.MimeTypeUtils;
 public class SenderService {
 
 	@Autowired
-	private Source source;
+	private MySource source;
 
 	public void send(String msg) throws Exception {
-		source.output().send(MessageBuilder.withPayload(msg).build());
+		source.output1().send(MessageBuilder.withPayload(msg).build());
 	}
 
 	public <T> void sendWithTags(T msg, String tag) throws Exception {
 		Map<String, Object> map = new HashMap<>();
 		map.put(MessageConst.PROPERTY_TAGS, tag);
 		Message message = MessageBuilder.createMessage(msg, new MessageHeaders(map));
-		source.output().send(message);
+		source.output1().send(message);
 	}
 
 	public <T> void sendObject(T msg, String tag) throws Exception {
@@ -37,7 +37,17 @@ public class SenderService {
 				.setHeader(MessageConst.PROPERTY_TAGS, tag)
 				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
 				.build();
-		source.output().send(message);
+		source.output1().send(message);
+	}
+
+	public <T> void sendTransactionalMsg(T msg, boolean error) throws Exception {
+		MessageBuilder builder = MessageBuilder.withPayload(msg)
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
+		if (error) {
+			builder.setHeader("test", "1");
+		}
+		Message message = builder.build();
+		source.output2().send(message);
 	}
 
 }

spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties

@@ -2,8 +2,14 @@ spring.cloud.stream.default-binder=rocketmq
 
 spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876
 
-spring.cloud.stream.bindings.output.destination=test-topic
+spring.cloud.stream.bindings.output1.destination=test-topic
-spring.cloud.stream.bindings.output.content-type=application/json
+spring.cloud.stream.bindings.output1.content-type=application/json
+
+spring.cloud.stream.bindings.output2.destination=TransactionTopic
+spring.cloud.stream.bindings.output2.content-type=application/json
+spring.cloud.stream.rocketmq.bindings.output2.producer.transactional=true
+spring.cloud.stream.rocketmq.bindings.output2.producer.executer=org.springframework.cloud.alibaba.cloud.examples.MyTransactionExecuter
+spring.cloud.stream.rocketmq.bindings.output2.producer.transaction-check-listener=org.springframework.cloud.alibaba.cloud.examples.MyTransactionCheckListener
 
 spring.cloud.stream.bindings.input1.destination=test-topic
 spring.cloud.stream.bindings.input1.content-type=text/plain
@@ -26,5 +32,12 @@ spring.cloud.stream.rocketmq.bindings.input3.consumer.tags=tagObj
 spring.cloud.stream.bindings.input3.consumer.concurrency=20
 spring.cloud.stream.bindings.input3.consumer.maxAttempts=1
 
+spring.cloud.stream.bindings.input4.destination=TransactionTopic
+spring.cloud.stream.bindings.input4.content-type=text/plain
+spring.cloud.stream.bindings.input4.group=transaction-group
+spring.cloud.stream.bindings.input4.consumer.concurrency=210
+
+spring.application.name=rocketmq-example
+
 server.port=28081
 management.security.enabled=false

spring-cloud-alibaba-sentinel/src/main/java/org/springframework/cloud/alibaba/sentinel/custom/SentinelAutoConfiguration.java

@@ -19,7 +19,6 @@ package org.springframework.cloud.alibaba.sentinel.custom;
 import javax.annotation.PostConstruct;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -156,26 +155,22 @@ public class SentinelAutoConfiguration {
 	}
 
 	@Bean("sentinel-json-converter")
-	public JsonConverter jsonConverter(
+	public JsonConverter jsonConverter() {
-			@Qualifier("sentinel-object-mapper") ObjectMapper objectMapper) {
+		return new JsonConverter(objectMapper());
-		return new JsonConverter(objectMapper);
 	}
 
-	@Bean("sentinel-object-mapper")
+	private ObjectMapper objectMapper() {
-	public ObjectMapper objectMapper() {
 		return new ObjectMapper();
 	}
 
 	@ConditionalOnClass(XmlMapper.class)
 	protected static class SentinelXmlConfiguration {
 		@Bean("sentinel-xml-converter")
-		public XmlConverter xmlConverter(
+		public XmlConverter xmlConverter() {
-				@Qualifier("sentinel-xml-mapper") XmlMapper xmlMapper) {
+			return new XmlConverter(xmlMapper());
-			return new XmlConverter(xmlMapper);
 		}
 
-		@Bean("sentinel-xml-mapper")
+		private XmlMapper xmlMapper() {
-		public XmlMapper xmlMapper() {
 			return new XmlMapper();
 		}
 	}

spring-cloud-alibaba-sentinel/src/main/java/org/springframework/cloud/alibaba/sentinel/feign/SentinelFeign.java

@@ -89,13 +89,14 @@ public class SentinelFeign {
 					// check fallback and fallbackFactory properties
 					if (void.class != fallback) {
 						fallbackInstance = getFromContext(name, "fallback", fallback,
-								target);
+								target.type());
 						return new SentinelInvocationHandler(target, dispatch,
 								new FallbackFactory.Default(fallbackInstance));
 					}
 					if (void.class != fallbackFactory) {
 						fallbackFactoryInstance = (FallbackFactory) getFromContext(name,
-								"fallbackFactory", fallbackFactory, target);
+								"fallbackFactory", fallbackFactory,
+								FallbackFactory.class);
 						return new SentinelInvocationHandler(target, dispatch,
 								fallbackFactoryInstance);
 					}
@@ -103,7 +104,7 @@ public class SentinelFeign {
 				}
 
 				private Object getFromContext(String name, String type,
-						Class fallbackType, Target target) {
+						Class fallbackType, Class targetType) {
 					Object fallbackInstance = feignContext.getInstance(name,
 							fallbackType);
 					if (fallbackInstance == null) {
@@ -112,10 +113,10 @@ public class SentinelFeign {
 								type, fallbackType, name));
 					}
 
-					if (!target.type().isAssignableFrom(fallbackType)) {
+					if (!targetType.isAssignableFrom(fallbackType)) {
 						throw new IllegalStateException(String.format(
 								"Incompatible %s instance. Fallback/fallbackFactory of type %s is not assignable to %s for feign client %s",
-								type, fallbackType, target.type(), name));
+								type, fallbackType, targetType, name));
 					}
 					return fallbackInstance;
 				}

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java

@@ -21,7 +21,7 @@ package org.springframework.cloud.stream.binder.rocketmq;
  */
 public interface RocketMQBinderConstants {
 
-    String ENDPOINT_ID = "rocketmq_binder";
+	String ENDPOINT_ID = "rocketmq_binder";
 
 	/**
 	 * Header key
@@ -32,6 +32,8 @@ public interface RocketMQBinderConstants {
 
 	String ROCKET_SEND_RESULT = "ROCKETMQ_SEND_RESULT";
 
+	String ROCKET_TRANSACTIONAL_ARG = "ROCKETMQ_TRANSACTIONAL_ARG";
+
 	String ACKNOWLEDGEMENT_KEY = "ACKNOWLEDGEMENT";
 
 	/**
@@ -39,23 +41,23 @@ public interface RocketMQBinderConstants {
 	 */
 	String LASTSEND_TIMESTAMP = "lastSend.timestamp";
 
-    interface Metrics {
+	interface Metrics {
-        interface Producer {
+		interface Producer {
-            String PREFIX = "scs-rocketmq.producer.";
+			String PREFIX = "scs-rocketmq.producer.";
-            String TOTAL_SENT = "totalSent";
+			String TOTAL_SENT = "totalSent";
-            String TOTAL_SENT_FAILURES = "totalSentFailures";
+			String TOTAL_SENT_FAILURES = "totalSentFailures";
-            String SENT_PER_SECOND = "sentPerSecond";
+			String SENT_PER_SECOND = "sentPerSecond";
-            String SENT_FAILURES_PER_SECOND = "sentFailuresPerSecond";
+			String SENT_FAILURES_PER_SECOND = "sentFailuresPerSecond";
-        }
+		}
 
-        interface Consumer {
+		interface Consumer {
-            String GROUP_PREFIX = "scs-rocketmq.consumerGroup.";
+			String GROUP_PREFIX = "scs-rocketmq.consumerGroup.";
-            String PREFIX = "scs-rocketmq.consumer.";
+			String PREFIX = "scs-rocketmq.consumer.";
-            String TOTAL_CONSUMED = "totalConsumed";
+			String TOTAL_CONSUMED = "totalConsumed";
-            String CONSUMED_PER_SECOND = "consumedPerSecond";
+			String CONSUMED_PER_SECOND = "consumedPerSecond";
-            String TOTAL_CONSUMED_FAILURES = "totalConsumedFailures";
+			String TOTAL_CONSUMED_FAILURES = "totalConsumedFailures";
-            String CONSUMED_FAILURES_PER_SECOND = "consumedFailuresPerSecond";
+			String CONSUMED_FAILURES_PER_SECOND = "consumedFailuresPerSecond";
-        }
+		}
-    }
+	}
 
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java

@@ -16,6 +16,9 @@
 
 package org.springframework.cloud.stream.binder.rocketmq;
 
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
+import org.apache.rocketmq.client.producer.TransactionCheckListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
@@ -36,6 +39,7 @@ import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.util.ClassUtils;
 
 /**
  * @author Timur Valiev
@@ -71,9 +75,24 @@ public class RocketMQMessageChannelBinder extends
 			ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
 			MessageChannel errorChannel) throws Exception {
 		if (producerProperties.getExtension().getEnabled()) {
-			return new RocketMQMessageHandler(destination.getName(),
+			RocketMQMessageHandler messageHandler = new RocketMQMessageHandler(
-					producerProperties.getExtension(),
+					destination.getName(), producerProperties,
 					rocketBinderConfigurationProperties, instrumentationManager);
+			if (producerProperties.getExtension().getTransactional()) {
+				// transaction message check LocalTransactionExecuter
+				messageHandler.setLocalTransactionExecuter(
+						getClassConfiguration(destination.getName(),
+								producerProperties.getExtension().getExecuter(),
+								LocalTransactionExecuter.class));
+				// transaction message check TransactionCheckListener
+				messageHandler.setTransactionCheckListener(
+						getClassConfiguration(destination.getName(),
+								producerProperties.getExtension()
+										.getTransactionCheckListener(),
+								TransactionCheckListener.class));
+			}
+
+			return messageHandler;
 		}
 		else {
 			throw new RuntimeException("Binding for channel " + destination.getName()
@@ -120,4 +139,46 @@ public class RocketMQMessageChannelBinder extends
 	public RocketMQProducerProperties getExtendedProducerProperties(String channelName) {
 		return extendedBindingProperties.getExtendedProducerProperties(channelName);
 	}
+
+	private <T> T getClassConfiguration(String destName, String className,
+			Class<T> interfaceClass) {
+		if (StringUtils.isEmpty(className)) {
+			throw new RuntimeException("Binding for channel " + destName
+					+ " using transactional message, should set "
+					+ interfaceClass.getSimpleName() + " configuration"
+					+ interfaceClass.getSimpleName() + " should be set, like "
+					+ "'spring.cloud.stream.rocketmq.bindings.output.producer.xxx=TheFullClassNameOfYour"
+					+ interfaceClass.getSimpleName() + "'");
+		}
+		else if (StringUtils.isNotEmpty(className)) {
+			Class fieldClass;
+			// check class exists
+			try {
+				fieldClass = ClassUtils.forName(className,
+						RocketMQMessageChannelBinder.class.getClassLoader());
+			}
+			catch (ClassNotFoundException e) {
+				throw new RuntimeException("Binding for channel " + destName
+						+ " using transactional message, but " + className
+						+ " class is not found");
+			}
+			// check interface incompatible
+			if (!interfaceClass.isAssignableFrom(fieldClass)) {
+				throw new RuntimeException("Binding for channel " + destName
+						+ " using transactional message, but " + className
+						+ " is incompatible with " + interfaceClass.getSimpleName()
+						+ " interface");
+			}
+			try {
+				return (T) fieldClass.newInstance();
+			}
+			catch (Exception e) {
+				throw new RuntimeException("Binding for channel " + destName
+						+ " using transactional message, but " + className
+						+ " instance error", e);
+			}
+		}
+		return null;
+	}
+
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java

@@ -16,6 +16,12 @@
 
 package org.springframework.cloud.stream.binder.rocketmq;
 
+import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ACKNOWLEDGEMENT_KEY;
+import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ORIGINAL_ROCKET_MESSAGE;
+import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_FLAG;
+import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_SEND_RESULT;
+import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_TRANSACTIONAL_ARG;
+
 import java.util.HashMap;
 import java.util.Map;
 
@@ -29,95 +35,105 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.MessageHeaderAccessor;
 
-import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ACKNOWLEDGEMENT_KEY;
-import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ORIGINAL_ROCKET_MESSAGE;
-import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_FLAG;
-import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_SEND_RESULT;
-
 /**
  * @author Timur Valiev
  * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
  */
 public class RocketMQMessageHeaderAccessor extends MessageHeaderAccessor {
 
-    public RocketMQMessageHeaderAccessor() {
+	public RocketMQMessageHeaderAccessor() {
-        super();
+		super();
-    }
+	}
 
-    public RocketMQMessageHeaderAccessor(Message<?> message) {
+	public RocketMQMessageHeaderAccessor(Message<?> message) {
-        super(message);
+		super(message);
-    }
+	}
 
-    public Acknowledgement getAcknowledgement(Message message) {
+	public Acknowledgement getAcknowledgement(Message message) {
-        return message.getHeaders().get(ACKNOWLEDGEMENT_KEY, Acknowledgement.class);
+		return message.getHeaders().get(ACKNOWLEDGEMENT_KEY, Acknowledgement.class);
-    }
+	}
 
-    public RocketMQMessageHeaderAccessor withAcknowledgment(Acknowledgement acknowledgment) {
+	public RocketMQMessageHeaderAccessor withAcknowledgment(
-        setHeader(ACKNOWLEDGEMENT_KEY, acknowledgment);
+			Acknowledgement acknowledgment) {
-        return this;
+		setHeader(ACKNOWLEDGEMENT_KEY, acknowledgment);
-    }
+		return this;
+	}
 
-    public String getTags() {
+	public String getTags() {
-        return getMessageHeaders().get(MessageConst.PROPERTY_TAGS) == null ? "" : (String) getMessageHeaders().get(MessageConst.PROPERTY_TAGS);
+		return getMessageHeaders().get(MessageConst.PROPERTY_TAGS) == null ? ""
-    }
+				: (String) getMessageHeaders().get(MessageConst.PROPERTY_TAGS);
+	}
 
-    public RocketMQMessageHeaderAccessor withTags(String tag) {
+	public RocketMQMessageHeaderAccessor withTags(String tag) {
-        setHeader(MessageConst.PROPERTY_TAGS, tag);
+		setHeader(MessageConst.PROPERTY_TAGS, tag);
-        return this;
+		return this;
-    }
+	}
 
-    public String getKeys() {
+	public String getKeys() {
-        return getMessageHeaders().get(MessageConst.PROPERTY_KEYS) == null ? "" : (String) getMessageHeaders().get(MessageConst.PROPERTY_KEYS);
+		return getMessageHeaders().get(MessageConst.PROPERTY_KEYS) == null ? ""
-    }
+				: (String) getMessageHeaders().get(MessageConst.PROPERTY_KEYS);
+	}
 
-    public RocketMQMessageHeaderAccessor withKeys(String keys) {
+	public RocketMQMessageHeaderAccessor withKeys(String keys) {
-        setHeader(MessageConst.PROPERTY_KEYS, keys);
+		setHeader(MessageConst.PROPERTY_KEYS, keys);
-        return this;
+		return this;
-    }
+	}
 
-    public MessageExt getRocketMessage() {
+	public MessageExt getRocketMessage() {
-        return getMessageHeaders().get(ORIGINAL_ROCKET_MESSAGE, MessageExt.class);
+		return getMessageHeaders().get(ORIGINAL_ROCKET_MESSAGE, MessageExt.class);
-    }
+	}
 
-    public RocketMQMessageHeaderAccessor withRocketMessage(MessageExt message) {
+	public RocketMQMessageHeaderAccessor withRocketMessage(MessageExt message) {
-        setHeader(ORIGINAL_ROCKET_MESSAGE, message);
+		setHeader(ORIGINAL_ROCKET_MESSAGE, message);
-        return this;
+		return this;
-    }
+	}
 
-    public Integer getDelayTimeLevel() {
+	public Integer getDelayTimeLevel() {
-        return NumberUtils.toInt((String)getMessageHeaders().get(MessageConst.PROPERTY_DELAY_TIME_LEVEL), 0);
+		return NumberUtils.toInt(
-    }
+				(String) getMessageHeaders().get(MessageConst.PROPERTY_DELAY_TIME_LEVEL),
+				0);
+	}
 
-    public RocketMQMessageHeaderAccessor withDelayTimeLevel(Integer delayTimeLevel) {
+	public RocketMQMessageHeaderAccessor withDelayTimeLevel(Integer delayTimeLevel) {
-        setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel);
+		setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel);
-        return this;
+		return this;
-    }
+	}
 
-    public Integer getFlag() {
+	public Integer getFlag() {
-        return NumberUtils.toInt((String)getMessageHeaders().get(ROCKET_FLAG), 0);
+		return NumberUtils.toInt((String) getMessageHeaders().get(ROCKET_FLAG), 0);
-    }
+	}
 
-    public RocketMQMessageHeaderAccessor withFlag(Integer delayTimeLevel) {
+	public RocketMQMessageHeaderAccessor withFlag(Integer delayTimeLevel) {
-        setHeader(ROCKET_FLAG, delayTimeLevel);
+		setHeader(ROCKET_FLAG, delayTimeLevel);
-        return this;
+		return this;
-    }
+	}
 
-    public SendResult getSendResult() {
+	public Object getTransactionalArg() {
-        return getMessageHeaders().get(ROCKET_SEND_RESULT, SendResult.class);
+		return getMessageHeaders().get(ROCKET_TRANSACTIONAL_ARG);
-    }
+	}
 
-    public static void putSendResult(MutableMessage message, SendResult sendResult) {
+	public Object withTransactionalArg(Object arg) {
-        message.getHeaders().put(ROCKET_SEND_RESULT, sendResult);
+		setHeader(ROCKET_TRANSACTIONAL_ARG, arg);
-    }
+		return this;
+	}
 
-    public Map<String, String> getUserProperties() {
+	public SendResult getSendResult() {
-        Map<String, String> result = new HashMap<>();
+		return getMessageHeaders().get(ROCKET_SEND_RESULT, SendResult.class);
-        for (Map.Entry<String, Object> entry : this.toMap().entrySet()) {
+	}
-            if (entry.getValue() instanceof String && !MessageConst.STRING_HASH_SET.contains(entry.getKey()) && !entry
+
-                .getKey().equals(MessageHeaders.CONTENT_TYPE)) {
+	public static void putSendResult(MutableMessage message, SendResult sendResult) {
-                result.put(entry.getKey(), (String)entry.getValue());
+		message.getHeaders().put(ROCKET_SEND_RESULT, sendResult);
-            }
+	}
-        }
+
-        return result;
+	public Map<String, String> getUserProperties() {
-    }
+		Map<String, String> result = new HashMap<>();
+		for (Map.Entry<String, Object> entry : this.toMap().entrySet()) {
+			if (entry.getValue() instanceof String
+					&& !MessageConst.STRING_HASH_SET.contains(entry.getKey())
+					&& !entry.getKey().equals(MessageHeaders.CONTENT_TYPE)) {
+				result.put(entry.getKey(), (String) entry.getValue());
+			}
+		}
+		return result;
+	}
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java

@@ -21,10 +21,14 @@ import java.util.Map;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.client.producer.TransactionCheckListener;
+import org.apache.rocketmq.client.producer.TransactionMQProducer;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
 import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageHeaderAccessor;
 import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
@@ -47,16 +51,20 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 
 	private InstrumentationManager instrumentationManager;
 
-	private final RocketMQProducerProperties producerProperties;
+	private LocalTransactionExecuter localTransactionExecuter;
+
+	private TransactionCheckListener transactionCheckListener;
+
+	private final ExtendedProducerProperties<RocketMQProducerProperties> producerProperties;
 
 	private final String destination;
 
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 
-	protected volatile boolean running = false;
+	private volatile boolean running = false;
 
 	public RocketMQMessageHandler(String destination,
-			RocketMQProducerProperties producerProperties,
+			ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
 			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
 			InstrumentationManager instrumentationManager) {
 		this.destination = destination;
@@ -67,7 +75,16 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 
 	@Override
 	public void start() {
-		producer = new DefaultMQProducer(destination);
+		if (producerProperties.getExtension().getTransactional()) {
+			producer = new TransactionMQProducer(destination);
+			if (transactionCheckListener != null) {
+				((TransactionMQProducer) producer)
+						.setTransactionCheckListener(transactionCheckListener);
+			}
+		}
+		else {
+			producer = new DefaultMQProducer(destination);
+		}
 
 		if (instrumentationManager != null) {
 			producerInstrumentation = instrumentationManager
@@ -77,8 +94,9 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 
 		producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
 
-		if (producerProperties.getMaxMessageSize() > 0) {
+		if (producerProperties.getExtension().getMaxMessageSize() > 0) {
-			producer.setMaxMessageSize(producerProperties.getMaxMessageSize());
+			producer.setMaxMessageSize(
+					producerProperties.getExtension().getMaxMessageSize());
 		}
 
 		try {
@@ -139,7 +157,14 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 				toSend.putUserProperty(entry.getKey(), entry.getValue());
 			}
 
-			SendResult sendRes = producer.send(toSend);
+			SendResult sendRes;
+			if (producerProperties.getExtension().getTransactional()) {
+				sendRes = producer.sendMessageInTransaction(toSend,
+						localTransactionExecuter, headerAccessor.getTransactionalArg());
+			}
+			else {
+				sendRes = producer.send(toSend);
+			}
 
 			if (!sendRes.getSendStatus().equals(SendStatus.SEND_OK)) {
 				throw new MQClientException("message hasn't been sent", null);
@@ -167,4 +192,14 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 
 	}
 
+	public void setLocalTransactionExecuter(
+			LocalTransactionExecuter localTransactionExecuter) {
+		this.localTransactionExecuter = localTransactionExecuter;
+	}
+
+	public void setTransactionCheckListener(
+			TransactionCheckListener transactionCheckListener) {
+		this.transactionCheckListener = transactionCheckListener;
+	}
+
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java

@@ -17,6 +17,8 @@
 package org.springframework.cloud.stream.binder.rocketmq.properties;
 
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
+import org.apache.rocketmq.client.producer.TransactionCheckListener;
 
 /**
  * @author Timur Valiev
@@ -24,28 +26,62 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
  */
 public class RocketMQProducerProperties {
 
-    private Boolean enabled = true;
+	private Boolean enabled = true;
 
-    /**
+	/**
-     * Maximum allowed message size in bytes
+	 * Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}
-     * {@link DefaultMQProducer#maxMessageSize}
+	 */
-     */
+	private Integer maxMessageSize = 0;
-    private Integer maxMessageSize = 0;
 
-    public Boolean getEnabled() {
+	private Boolean transactional = false;
-        return enabled;
-    }
 
-    public void setEnabled(Boolean enabled) {
+	/**
-        this.enabled = enabled;
+	 * full class name of {@link LocalTransactionExecuter}
-    }
+	 */
+	private String executer;
 
-    public Integer getMaxMessageSize() {
+	/**
-        return maxMessageSize;
+	 * full class name of {@link TransactionCheckListener}
-    }
+	 */
+	private String transactionCheckListener;
 
-    public void setMaxMessageSize(Integer maxMessageSize) {
+	public Boolean getEnabled() {
-        this.maxMessageSize = maxMessageSize;
+		return enabled;
-    }
+	}
 
+	public void setEnabled(Boolean enabled) {
+		this.enabled = enabled;
+	}
+
+	public Integer getMaxMessageSize() {
+		return maxMessageSize;
+	}
+
+	public void setMaxMessageSize(Integer maxMessageSize) {
+		this.maxMessageSize = maxMessageSize;
+	}
+
+	public Boolean getTransactional() {
+		return transactional;
+	}
+
+	public void setTransactional(Boolean transactional) {
+		this.transactional = transactional;
+	}
+
+	public String getExecuter() {
+		return executer;
+	}
+
+	public void setExecuter(String executer) {
+		this.executer = executer;
+	}
+
+	public String getTransactionCheckListener() {
+		return transactionCheckListener;
+	}
+
+	public void setTransactionCheckListener(String transactionCheckListener) {
+		this.transactionCheckListener = transactionCheckListener;
+	}
 }