diff --git a/cdp-common/cdp-common-core/pom.xml b/cdp-common/cdp-common-core/pom.xml
index 974cf7d..304c7a0 100644
--- a/cdp-common/cdp-common-core/pom.xml
+++ b/cdp-common/cdp-common-core/pom.xml
@@ -12,8 +12,8 @@
cdp-common-core
- 11
- 11
+ 8
+ 8
diff --git a/cdp-common/cdp-common-rocketmq/pom.xml b/cdp-common/cdp-common-rocketmq/pom.xml
new file mode 100644
index 0000000..d6ae33d
--- /dev/null
+++ b/cdp-common/cdp-common-rocketmq/pom.xml
@@ -0,0 +1,51 @@
+
+
+ 4.0.0
+
+ cdp-common
+ com.baiye
+ 1.0-SNAPSHOT
+
+ cdp-common-rocketmq
+ 1.0.0-SNAPSHOT
+ cdp-common-rocketmq
+ cdp-common-rocketmq
+
+ 1.8
+
+
+
+ com.baiye
+ cdp-common-core
+ 1.0-SNAPSHOT
+
+
+
+
+ com.alibaba.cloud
+ spring-cloud-starter-stream-rocketmq
+
+
+ org.springframework.integration
+ spring-integration-core
+
+
+
+
+
+ org.springframework.integration
+ spring-integration-core
+
+
+
+
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+
+
+
+
diff --git a/cdp-common/cdp-common-security/pom.xml b/cdp-common/cdp-common-security/pom.xml
index 41941a5..8bac6ea 100644
--- a/cdp-common/cdp-common-security/pom.xml
+++ b/cdp-common/cdp-common-security/pom.xml
@@ -12,8 +12,8 @@
cdp-common-security
- 11
- 11
+ 8
+ 8
diff --git a/cdp-common/pom.xml b/cdp-common/pom.xml
index 26e9737..f44e677 100644
--- a/cdp-common/pom.xml
+++ b/cdp-common/pom.xml
@@ -16,6 +16,7 @@
cdp-common-core
cdp-common-exception
cdp-common-mybatis-plus
+ cdp-common-rocketmq
diff --git a/cdp-tools/cdp-tool-mq/.gitignore b/cdp-tools/cdp-tool-mq/.gitignore
new file mode 100644
index 0000000..c84de1f
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/.gitignore
@@ -0,0 +1,34 @@
+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/
+.mvn
diff --git a/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/pom.xml b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/pom.xml
new file mode 100644
index 0000000..dbf03ca
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/pom.xml
@@ -0,0 +1,36 @@
+
+
+ 4.0.0
+
+ cdp-tool-mq
+ com.baiye
+ 1.0.0-SNAPSHOT
+
+ cdp-tool-mq-producer
+ 1.0.0-SNAPSHOT
+ cdp-tool-mq-producer
+ cdp-tool-mq-producer
+
+
+ 1.8
+
+
+
+
+ com.baiye
+ cdp-common-rocketmq
+ 1.0.0-SNAPSHOT
+
+
+
+
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+
+
+
+
diff --git a/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/CdpToolMqProducerApplication.java b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/CdpToolMqProducerApplication.java
new file mode 100644
index 0000000..4de52d2
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/CdpToolMqProducerApplication.java
@@ -0,0 +1,22 @@
+package com.baiye;
+
+import com.baiye.message.MySource;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+
+/**
+ * @author Enzo
+ * @date 2022-7-13
+ */
+
+@EnableBinding(MySource.class)
+@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})
+public class CdpToolMqProducerApplication {
+
+ public static void main(String[] args) {
+ SpringApplication.run(CdpToolMqProducerApplication.class, args);
+ }
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/controller/Demo01Controller.java b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/controller/Demo01Controller.java
new file mode 100644
index 0000000..1eca649
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/controller/Demo01Controller.java
@@ -0,0 +1,42 @@
+package com.baiye.controller;
+
+import com.baiye.message.Demo01Message;
+import com.baiye.message.MySource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.annotation.Resource;
+import java.util.Random;
+
+@RestController
+@RequestMapping("/demo01")
+public class Demo01Controller {
+
+ private Logger logger = LoggerFactory.getLogger(getClass());
+
+ @Resource
+ private MySource mySource;
+
+ @GetMapping("/send_orderly")
+ public boolean sendOrderly() {
+ // 发送 3 条相同 id 的消息
+ int id = new Random().nextInt();
+ for (int i = 0; i < 3; i++) {
+ // 创建 Message
+ Demo01Message message = new Demo01Message().setId(id);
+ // 创建 Spring Message 对象
+ Message springMessage = MessageBuilder.withPayload(message)
+ .build();
+ // 发送消息
+ mySource.testOutPut().send(springMessage);
+ }
+ return true;
+ }
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/message/Demo01Message.java b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/message/Demo01Message.java
new file mode 100644
index 0000000..5bed8db
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/message/Demo01Message.java
@@ -0,0 +1,29 @@
+package com.baiye.message;
+
+/**
+ * 示例 01 的 Message 消息
+ */
+public class Demo01Message {
+
+ /**
+ * 编号
+ */
+ private Integer id;
+
+ public Demo01Message setId(Integer id) {
+ this.id = id;
+ return this;
+ }
+
+ public Integer getId() {
+ return id;
+ }
+
+ @Override
+ public String toString() {
+ return "Demo01Message{" +
+ "id=" + id +
+ '}';
+ }
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/message/MySource.java b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/message/MySource.java
new file mode 100644
index 0000000..cb16dd3
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/message/MySource.java
@@ -0,0 +1,11 @@
+package com.baiye.message;
+
+import org.springframework.cloud.stream.annotation.Output;
+import org.springframework.messaging.MessageChannel;
+
+public interface MySource {
+
+ @Output("test-output")
+ MessageChannel testOutPut();
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/message/TransactionListenerImpl.java b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/message/TransactionListenerImpl.java
new file mode 100644
index 0000000..ab2668d
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/java/com/baiye/message/TransactionListenerImpl.java
@@ -0,0 +1,35 @@
+package com.baiye.message;
+
+import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.messaging.Message;
+
+/**
+ * @description 实现 MQ 事务的监听。
+ * @ClassName: TransactionListenerImpl
+ * @author: 郭秀志 jbcode@126.com
+ * @date: 2020/8/7 9:21
+ * @Copyright:
+ */
+@RocketMQTransactionListener(txProducerGroup = "test")
+public class TransactionListenerImpl implements RocketMQLocalTransactionListener {
+
+ private Logger logger = LoggerFactory.getLogger(getClass());
+
+ @Override
+ public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
+
+ return RocketMQLocalTransactionState.UNKNOWN;
+ }
+
+ @Override
+ public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
+ // ... check local transaction status and return rollback, commit or unknown
+ logger.info("[checkLocalTransaction][回查消息:{}]", msg);
+ return RocketMQLocalTransactionState.COMMIT;
+ }
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/resources/application.yml b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/resources/application.yml
new file mode 100644
index 0000000..ec039d2
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/main/resources/application.yml
@@ -0,0 +1,31 @@
+spring:
+ application:
+ name: stream-rocketmq-producer-application
+ cloud:
+ # Spring Cloud Stream 配置项,对应 BindingServiceProperties 类
+ stream:
+ # Binding 配置项,对应 BindingProperties Map
+ bindings:
+ erbadagang-output:
+ destination: TEST-TOPIC-01 # 目的地。这里使用 RocketMQ Topic
+ content-type: application/json # 内容格式。这里使用 JSON
+ # Producer 配置项,对应 ProducerProperties 类
+ producer:
+ partition-key-expression: payload['id'] # 分区 key 表达式。该表达式基于 Spring EL,从消息中获得分区 key。
+
+ # Spring Cloud Stream RocketMQ 配置项
+ rocketmq:
+ # RocketMQ Binder 配置项,对应 RocketMQBinderConfigurationProperties 类
+ binder:
+ name-server: 8.130.96.163:9876 # RocketMQ Namesrv 地址
+ # RocketMQ 自定义 Binding 配置项,对应 RocketMQBindingProperties Map
+ bindings:
+ test-output:
+ # RocketMQ Producer 配置项,对应 RocketMQProducerProperties 类
+ producer:
+ transactional: true
+ group: test # 生产者分组
+ sync: true # 是否同步发送消息,默认为 false 异步。
+
+server:
+ port: 18080
diff --git a/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/test/java/com/baiye/CdpToolMqProducerApplicationTests.java b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/test/java/com/baiye/CdpToolMqProducerApplicationTests.java
new file mode 100644
index 0000000..75172fc
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cdp-tool-mq-producer/src/test/java/com/baiye/CdpToolMqProducerApplicationTests.java
@@ -0,0 +1,13 @@
+package com.baiye;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class CdpToolMqProducerApplicationTests {
+
+ @Test
+ void contextLoads() {
+ }
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/pom.xml b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/pom.xml
new file mode 100644
index 0000000..a02c041
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/pom.xml
@@ -0,0 +1,37 @@
+
+
+ 4.0.0
+
+ org.springframework.boot
+ spring-boot-starter-parent
+ 2.7.1
+
+
+ com.baiye
+ cpd-tool-mq-consumer
+ 0.0.1-SNAPSHOT
+ cpd-tool-mq-consumer
+ cpd-tool-mq-consumer
+
+ 1.8
+
+
+
+
+ com.baiye
+ cdp-common-rocketmq
+ 1.0.0-SNAPSHOT
+
+
+
+
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+
+
+
+
diff --git a/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/CpdToolMqConsumerApplication.java b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/CpdToolMqConsumerApplication.java
new file mode 100644
index 0000000..2e1739e
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/CpdToolMqConsumerApplication.java
@@ -0,0 +1,22 @@
+package com.baiye;
+
+import com.baiye.listener.MySink;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+
+/**
+ * @author Enzo
+ * @date 2022-7-13
+ */
+
+@EnableBinding(MySink.class)
+@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})
+public class CpdToolMqConsumerApplication {
+
+ public static void main(String[] args) {
+ SpringApplication.run(CpdToolMqConsumerApplication.class, args);
+ }
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/listener/Demo01Consumer.java b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/listener/Demo01Consumer.java
new file mode 100644
index 0000000..63e07ba
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/listener/Demo01Consumer.java
@@ -0,0 +1,20 @@
+package com.baiye.listener;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.messaging.Message;
+import org.springframework.stereotype.Component;
+
+@Component
+public class Demo01Consumer {
+
+ private Logger logger = LoggerFactory.getLogger(getClass());
+
+ @StreamListener(MySink.TEST_INPUT)
+ public void onMessage(Message> message) {
+ logger.info("[onMessage][线程编号:{} 消息内容:{}]", Thread.currentThread().getId(), message);
+ // throw new RuntimeException("抛出一个异常");
+ }
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/listener/MySink.java b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/listener/MySink.java
new file mode 100644
index 0000000..4a8fb85
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/listener/MySink.java
@@ -0,0 +1,13 @@
+package com.baiye.listener;
+
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.messaging.SubscribableChannel;
+
+public interface MySink {
+
+ String TEST_INPUT = "test-input";
+
+ @Input(TEST_INPUT)
+ SubscribableChannel demo01Input();
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/message/Demo01Message.java b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/message/Demo01Message.java
new file mode 100644
index 0000000..5bed8db
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/java/com/baiye/message/Demo01Message.java
@@ -0,0 +1,29 @@
+package com.baiye.message;
+
+/**
+ * 示例 01 的 Message 消息
+ */
+public class Demo01Message {
+
+ /**
+ * 编号
+ */
+ private Integer id;
+
+ public Demo01Message setId(Integer id) {
+ this.id = id;
+ return this;
+ }
+
+ public Integer getId() {
+ return id;
+ }
+
+ @Override
+ public String toString() {
+ return "Demo01Message{" +
+ "id=" + id +
+ '}';
+ }
+
+}
diff --git a/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/resources/application.yml b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/resources/application.yml
new file mode 100644
index 0000000..43c1f9f
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/main/resources/application.yml
@@ -0,0 +1,31 @@
+
+spring:
+ application:
+ name: @artifactId@
+ cloud:
+ # Spring Cloud Stream ?????? BindingServiceProperties ?
+ stream:
+ # Binding ?????? BindingProperties Map
+ bindings:
+ test-input:
+ destination: TEST-TOPIC-01 # ???????? RocketMQ Topic
+ content-type: application/json # ????????? JSON
+ group: test-consumer-group-test-topic-01 # ?????,???????+topic?
+
+ # Spring Cloud Stream RocketMQ ???
+ rocketmq:
+ # RocketMQ Binder ?????? RocketMQBinderConfigurationProperties ?
+ binder:
+ name-server: 8.130.96.163:9876 # RocketMQ Namesrv ??
+ # RocketMQ ??? Binding ?????? RocketMQBindingProperties Map
+ bindings:
+ test-input:
+ # RocketMQ Consumer ?????? RocketMQConsumerProperties ?
+ consumer:
+ delay-level-when-next-consume: 0 # ????????????????????? 0
+ enabled: true # ?????????? true
+ broadcasting: false # ???????????? false ??????
+ orderly: true # ?????????? false ?????
+
+server:
+ port: ${random.int[10000,19999]} # ??????????????
diff --git a/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/test/java/com/baiye/CpdToolMqConsumerApplicationTests.java b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/test/java/com/baiye/CpdToolMqConsumerApplicationTests.java
new file mode 100644
index 0000000..8613e94
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/cpd-tool-mq-consumer/src/test/java/com/baiye/CpdToolMqConsumerApplicationTests.java
@@ -0,0 +1,13 @@
+package com.baiye;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class CpdToolMqConsumerApplicationTests {
+
+ @Test
+ void contextLoads() {
+ }
+
+}
diff --git a/cdp-tools/cdp-tool-mq/pom.xml b/cdp-tools/cdp-tool-mq/pom.xml
new file mode 100644
index 0000000..9026b92
--- /dev/null
+++ b/cdp-tools/cdp-tool-mq/pom.xml
@@ -0,0 +1,25 @@
+
+
+ 4.0.0
+
+ cdp-tools
+ com.baiye
+ 1.0-SNAPSHOT
+
+ cdp-tool-mq
+ 1.0.0-SNAPSHOT
+ cdp-tool-mq
+ cdp-tool-mq
+ pom
+
+ cdp-tool-mq-producer
+ cpd-tool-mq-consumer
+
+
+ 1.8
+
+
+
+
+
diff --git a/cdp-tools/pom.xml b/cdp-tools/pom.xml
index 6c05e1e..865177a 100644
--- a/cdp-tools/pom.xml
+++ b/cdp-tools/pom.xml
@@ -15,6 +15,7 @@
dy-tool-livetelecast
dy-tool-member
dy-tool-video
+ cdp-tool-mq
@@ -22,4 +23,4 @@
8
-
\ No newline at end of file
+
diff --git a/pom.xml b/pom.xml
index eefef2d..b47d0ea 100644
--- a/pom.xml
+++ b/pom.xml
@@ -270,6 +270,13 @@
springfox-swagger2
${swagger2.version}
+
+
+ org.springframework.integration
+ spring-integration-core
+ 5.2.1.RELEASE
+
+