rocketmq example readme improve

spring-cloud-alibaba-examples/pom.xml

@@ -43,6 +43,7 @@
         <module>rocketmq-example/rocketmq-delay-consume-example</module>
         <module>rocketmq-example/rocketmq-sql-consume-example</module>
         <module>rocketmq-example/rocketmq-example-common</module>
+        <module>rocketmq-example/rocketmq-tx-example</module>
 
         <module>spring-cloud-bus-rocketmq-example</module>
         <module>spring-cloud-alibaba-dubbo-examples</module>

spring-cloud-alibaba-examples/rocketmq-example/readme-zh.md

@@ -681,6 +681,136 @@ public class RocketMQSqlConsumeApplication {
 }
 ```
 
+## 事务消息示例
+
+### 什么是事务消息?
+
+参考[Transaction Example](https://rocketmq.apache.org/docs/transaction-example/).
+
+> 可以被认为是一个两阶段的提交消息实现，以确保分布式系统的最终一致性。 事务性消息确保本地事务的执行和消息的发送可以原子地执行。
+
+### Application
+
+> 1、 事务装填
+>
+> 事务消息有三个状态:
+> (1) TransactionStatus.CommitTransaction: 提交事务，意味着消费者可以消费事务
+> (2) TransactionStatus.RollbackTransaction: 回滚事务，消息将被删除，并且不允许被消费。
+> (3) TransactionStatus.Unknown: 中间状态，意味着MQ需要回查最终状态。
+
+### 创建Topic
+
+```sh
+sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t tx
+```
+
+### 示例代码
+
+**application.yml**
+
+```yaml
+server:
+  port: 28088
+spring:
+  application:
+    name: rocketmq-tx-example
+  cloud:
+    stream:
+      function:
+        definition: consumer;
+      rocketmq:
+        binder:
+          name-server: localhost:9876
+        bindings:
+          producer-out-0:
+            producer:
+              group: output_1
+              transactionListener: myTransactionListener
+              producerType: Trans
+      bindings:
+        producer-out-0:
+          destination: tx
+        consumer-in-0:
+          destination: tx
+          group: tx-group
+logging:
+  level:
+    org.springframework.context.support: debug
+```
+
+**TransactionListenerImpl**
+
+执行本地事务。
+
+```java
+@Component("myTransactionListener")
+public class TransactionListenerImpl implements TransactionListener {
+
+	@Override
+	public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
+		Object num = msg.getProperty("test");
+
+		if ("1".equals(num)) {
+			System.out.println("executer: " + new String(msg.getBody()) + " unknown");
+			return LocalTransactionState.UNKNOW;
+		}
+		else if ("2".equals(num)) {
+			System.out.println("executer: " + new String(msg.getBody()) + " rollback");
+			return LocalTransactionState.ROLLBACK_MESSAGE;
+		}
+		System.out.println("executer: " + new String(msg.getBody()) + " commit");
+		return LocalTransactionState.COMMIT_MESSAGE;
+	}
+
+	@Override
+	public LocalTransactionState checkLocalTransaction(MessageExt msg) {
+		System.out.println("check: " + new String(msg.getBody()));
+		return LocalTransactionState.COMMIT_MESSAGE;
+	}
+
+}
+```
+
+**producer and consumer**
+
+```java
+@SpringBootApplication
+public class RocketMQTxApplication {
+	private static final Logger log = LoggerFactory
+			.getLogger(RocketMQTxApplication.class);
+	@Autowired
+	private StreamBridge streamBridge;
+	public static void main(String[] args) {
+		SpringApplication.run(RocketMQTxApplication.class, args);
+	}
+
+
+	@Bean
+	public ApplicationRunner producer() {
+		return args -> {
+			for (int i = 1; i <= 4; i++) {
+				MessageBuilder builder = MessageBuilder.withPayload(new SimpleMsg("Hello Tx msg " + i));
+				builder.setHeader("test", String.valueOf(i))
+						.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
+				builder.setHeader(RocketMQConst.USER_TRANSACTIONAL_ARGS, "binder");
+				Message<SimpleMsg> msg = builder.build();
+				streamBridge.send("producer-out-0", msg);
+				System.out.println("send Msg:" + msg.toString());
+			}
+		};
+	}
+
+	@Bean
+	public Consumer<Message<SimpleMsg>> consumer() {
+		return msg -> {
+			Object arg = msg.getHeaders();
+			log.info(Thread.currentThread().getName() + " Receive New Messages: " + msg.getPayload().getMsg() + " ARG:"
+				+ arg.toString());
+		};
+	}
+}
+```
+
 ## Endpoint 信息查看
 
 Spring Boot 应用支持通过 Endpoint 来暴露相关信息，RocketMQ Stream Starter 也支持这一点。

spring-cloud-alibaba-examples/rocketmq-example/readme.md

@@ -675,6 +675,138 @@ public class RocketMQSqlConsumeApplication {
 }
 ```
 
+## Transaction example
+
+### What is transactional message?
+
+Refer to [Transaction Example](https://rocketmq.apache.org/docs/transaction-example/).
+
+> It can be thought of as a two-phase commit message implementation to ensure eventual consistency in distributed system. Transactional message ensures that the execution of local transaction and the sending of message can be performed atomically.
+
+### Application
+
+Refer to https://rocketmq.apache.org/
+
+> 1、 Transactional status
+>
+> There are three states for transactional message:
+> (1) TransactionStatus.CommitTransaction: commit transaction，it means that allow consumers to consume this message.
+> (2) TransactionStatus.RollbackTransaction: rollback transaction，it means that the message will be deleted and not allowed to consume.
+> (3) TransactionStatus.Unknown: intermediate state，it means that MQ is needed to check back to determine the status.
+
+### Create topic
+
+```sh
+sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t tx
+```
+
+### Example code
+
+**application.yml**
+
+```yaml
+server:
+  port: 28088
+spring:
+  application:
+    name: rocketmq-tx-example
+  cloud:
+    stream:
+      function:
+        definition: consumer;
+      rocketmq:
+        binder:
+          name-server: localhost:9876
+        bindings:
+          producer-out-0:
+            producer:
+              group: output_1
+              transactionListener: myTransactionListener
+              producerType: Trans
+      bindings:
+        producer-out-0:
+          destination: tx
+        consumer-in-0:
+          destination: tx
+          group: tx-group
+logging:
+  level:
+    org.springframework.context.support: debug
+```
+
+**TransactionListenerImpl**
+
+To execute local transaction.
+
+```java
+@Component("myTransactionListener")
+public class TransactionListenerImpl implements TransactionListener {
+
+	@Override
+	public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
+		Object num = msg.getProperty("test");
+
+		if ("1".equals(num)) {
+			System.out.println("executer: " + new String(msg.getBody()) + " unknown");
+			return LocalTransactionState.UNKNOW;
+		}
+		else if ("2".equals(num)) {
+			System.out.println("executer: " + new String(msg.getBody()) + " rollback");
+			return LocalTransactionState.ROLLBACK_MESSAGE;
+		}
+		System.out.println("executer: " + new String(msg.getBody()) + " commit");
+		return LocalTransactionState.COMMIT_MESSAGE;
+	}
+
+	@Override
+	public LocalTransactionState checkLocalTransaction(MessageExt msg) {
+		System.out.println("check: " + new String(msg.getBody()));
+		return LocalTransactionState.COMMIT_MESSAGE;
+	}
+
+}
+```
+
+**producer and consumer**
+
+```java
+@SpringBootApplication
+public class RocketMQTxApplication {
+	private static final Logger log = LoggerFactory
+			.getLogger(RocketMQTxApplication.class);
+	@Autowired
+	private StreamBridge streamBridge;
+	public static void main(String[] args) {
+		SpringApplication.run(RocketMQTxApplication.class, args);
+	}
+
+
+	@Bean
+	public ApplicationRunner producer() {
+		return args -> {
+			for (int i = 1; i <= 4; i++) {
+				MessageBuilder builder = MessageBuilder.withPayload(new SimpleMsg("Hello Tx msg " + i));
+				builder.setHeader("test", String.valueOf(i))
+						.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
+				builder.setHeader(RocketMQConst.USER_TRANSACTIONAL_ARGS, "binder");
+				Message<SimpleMsg> msg = builder.build();
+				streamBridge.send("producer-out-0", msg);
+				System.out.println("send Msg:" + msg.toString());
+			}
+		};
+	}
+
+	@Bean
+	public Consumer<Message<SimpleMsg>> consumer() {
+		return msg -> {
+			Object arg = msg.getHeaders();
+			log.info(Thread.currentThread().getName() + " Receive New Messages: " + msg.getPayload().getMsg() + " ARG:"
+				+ arg.toString());
+		};
+	}
+}
+```
+
 ## Endpoint
 
 Add dependency `spring-cloud-starter-stream-rocketmq` to your pom.xml file, and configure your endpoint security strategy.

spring-cloud-alibaba-examples/rocketmq-example/rocketmq-tx-example/pom.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+    <parent>
+        <groupId>com.alibaba.cloud</groupId>
+        <artifactId>spring-cloud-alibaba-examples</artifactId>
+        <version>${revision}</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+
+    <artifactId>rocketmq-tx-example</artifactId>
+    <name>Spring Cloud Starter Stream Alibaba RocketMQ Transaction message Example</name>
+    <description>Example demonstrating how to send and consume transaction messages in rocketmq</description>
+    <packaging>jar</packaging>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.cloud</groupId>
+            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-actuator</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.cloud</groupId>
+            <artifactId>rocketmq-example-common</artifactId>
+            <version>${revision}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-deploy-plugin</artifactId>
+                <version>${maven-deploy-plugin.version}</version>
+                <configuration>
+                    <skip>true</skip>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

spring-cloud-alibaba-examples/rocketmq-example/rocketmq-tx-example/src/main/java/com/alibaba/cloud/examples/tx/RocketMQTxApplication.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.examples.tx;
+
+import java.util.function.Consumer;
+
+import com.alibaba.cloud.examples.common.SimpleMsg;
+import com.alibaba.cloud.stream.binder.rocketmq.constant.RocketMQConst;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * @author sorie
+ */
+@SpringBootApplication
+public class RocketMQTxApplication {
+	private static final Logger log = LoggerFactory
+			.getLogger(RocketMQTxApplication.class);
+	@Autowired
+	private StreamBridge streamBridge;
+	public static void main(String[] args) {
+		SpringApplication.run(RocketMQTxApplication.class, args);
+	}
+
+
+	@Bean
+	public ApplicationRunner producer() {
+		return args -> {
+			for (int i = 1; i <= 4; i++) {
+				MessageBuilder builder = MessageBuilder.withPayload(new SimpleMsg("Hello Tx msg " + i));
+				builder.setHeader("test", String.valueOf(i))
+						.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
+				builder.setHeader(RocketMQConst.USER_TRANSACTIONAL_ARGS, "binder");
+				Message<SimpleMsg> msg = builder.build();
+				streamBridge.send("producer-out-0", msg);
+				System.out.println("send Msg:" + msg.toString());
+			}
+		};
+	}
+
+	@Bean
+	public Consumer<Message<SimpleMsg>> consumer() {
+		return msg -> {
+			Object arg = msg.getHeaders();
+			log.info(Thread.currentThread().getName() + " Receive New Messages: " + msg.getPayload().getMsg() + " ARG:"
+				+ arg.toString());
+		};
+	}
+}

spring-cloud-alibaba-examples/rocketmq-example/rocketmq-tx-example/src/main/java/com/alibaba/cloud/examples/tx/TransactionListenerImpl.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.cloud.examples.tx;
+
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.client.producer.TransactionListener;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+@Component("myTransactionListener")
+public class TransactionListenerImpl implements TransactionListener {
+
+	/**
+	 * Excute local transaction.
+	 * @param msg messages
+	 * @param arg message args
+	 * @return Transaction state
+	 */
+	@Override
+	public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
+		Object num = msg.getProperty("test");
+
+		if ("1".equals(num)) {
+			System.out.println("executer: " + new String(msg.getBody()) + " unknown");
+			return LocalTransactionState.UNKNOW;
+		}
+		else if ("2".equals(num)) {
+			System.out.println("executer: " + new String(msg.getBody()) + " rollback");
+			return LocalTransactionState.ROLLBACK_MESSAGE;
+		}
+		System.out.println("executer: " + new String(msg.getBody()) + " commit");
+		return LocalTransactionState.COMMIT_MESSAGE;
+	}
+
+	/**
+	 * MQ check back local transaction states.
+	 * @param msg messages
+	 * @return Transaction state
+	 */
+	@Override
+	public LocalTransactionState checkLocalTransaction(MessageExt msg) {
+		System.out.println("check: " + new String(msg.getBody()));
+		return LocalTransactionState.COMMIT_MESSAGE;
+	}
+
+}

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

@@ -0,0 +1,27 @@
+server:
+  port: 28088
+spring:
+  application:
+    name: rocketmq-tx-example
+  cloud:
+    stream:
+      function:
+        definition: consumer;
+      rocketmq:
+        binder:
+          name-server: localhost:9876
+        bindings:
+          producer-out-0:
+            producer:
+              group: output_1
+              transactionListener: myTransactionListener
+              producerType: Trans
+      bindings:
+        producer-out-0:
+          destination: tx
+        consumer-in-0:
+          destination: tx
+          group: tx-group
+logging:
+  level:
+    org.springframework.context.support: debug