!163 将spring-cloud-stream改为普通的mq依赖用法

* update: sky日志
* update: RocketMQ的集成方式
* feat:1. rabbit: 普通消息、延迟队列

pom.xml

@@ -45,6 +45,7 @@
         <mapstruct-plus.version>1.3.6</mapstruct-plus.version>
         <mapstruct-plus.lombok.version>0.2.0</mapstruct-plus.lombok.version>
         <justauth.version>1.16.6</justauth.version>
+        <rocketmq-version>2.3.0</rocketmq-version>
         <!-- 离线IP地址定位库 -->
         <ip2region.version>2.7.0</ip2region.version>
         <!-- 临时修复 fastjson 漏洞 -->
@@ -384,6 +385,13 @@
                 <version>${mapstruct-plus.version}</version>
             </dependency>
 
+            <!--消息队列-->
+            <dependency>
+                <groupId>org.apache.rocketmq</groupId>
+                <artifactId>rocketmq-spring-boot-starter</artifactId>
+                <version>${rocketmq-version}</version>
+            </dependency>
+
         </dependencies>
     </dependencyManagement>
 

ruoyi-example/pom.xml

@@ -10,7 +10,7 @@
 
     <modules>
         <module>ruoyi-demo</module>
-        <module>ruoyi-stream-mq</module>
+        <module>ruoyi-test-mq</module>
     </modules>
 
     <artifactId>ruoyi-example</artifactId>

ruoyi-example/ruoyi-stream-mq/src/main/java/org/dromara/stream/controller/TestMqController.java

@@ -1,60 +0,0 @@
-package org.dromara.stream.controller;
-
-import org.dromara.common.core.domain.R;
-import org.dromara.stream.mq.producer.DelayProducer;
-import org.dromara.stream.mq.producer.LogStreamProducer;
-import org.dromara.stream.mq.producer.TestStreamProducer;
-import lombok.AllArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
-
-/**
- * 测试mq
- */
-@Slf4j
-@RestController
-@AllArgsConstructor
-@RequestMapping("/test-mq")
-public class TestMqController {
-
-    private final DelayProducer delayProducer;
-    private final TestStreamProducer testStreamProducer;
-    private final LogStreamProducer logStreamProducer;
-
-    /**
-     * 发送消息Rabbitmq
-     *
-     * @param msg 消息内容
-     * @param delay 延时时间
-     */
-    @GetMapping("/sendRabbitmq")
-    public R<Void> sendRabbitmq(String msg, Long delay) {
-        delayProducer.sendMsg(msg, delay);
-        return R.ok();
-    }
-
-    /**
-     * 发送消息Rocketmq
-     *
-     * @param msg 消息内容
-     */
-    @GetMapping("/sendRocketmq")
-    public R<Void> sendRocketmq(String msg) {
-        testStreamProducer.streamTestMsg(msg);
-        return R.ok();
-    }
-
-    /**
-     * 发送消息Kafka
-     *
-     * @param msg 消息内容
-     */
-    @GetMapping("/sendKafka")
-    public R<Void> sendKafka(String msg) {
-        logStreamProducer.streamLogMsg(msg);
-        return R.ok();
-    }
-
-}

ruoyi-example/ruoyi-stream-mq/src/main/java/org/dromara/stream/mq/TestMessaging.java

@@ -1,20 +0,0 @@
-package org.dromara.stream.mq;
-
-import lombok.Data;
-import lombok.experimental.Accessors;
-
-/**
- * @author Lion Li
- */
-@Data
-@Accessors(chain = true)
-public class TestMessaging {
-    /**
-     * 消息id
-     */
-    private String msgId;
-    /**
-     * 消息内容
-     */
-    private String msgText;
-}

ruoyi-example/ruoyi-stream-mq/src/main/java/org/dromara/stream/mq/consumer/DelayConsumer.java

@@ -1,22 +0,0 @@
-package org.dromara.stream.mq.consumer;
-
-
-import org.dromara.stream.mq.TestMessaging;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.context.annotation.Bean;
-import org.springframework.stereotype.Component;
-
-import java.util.function.Consumer;
-
-@Slf4j
-@Component
-public class DelayConsumer {
-
-    @Bean
-    Consumer<TestMessaging> delay() {
-        log.info("初始化订阅");
-        return obj -> {
-            log.info("消息接收成功：" + obj);
-        };
-    }
-}

ruoyi-example/ruoyi-stream-mq/src/main/java/org/dromara/stream/mq/consumer/LogStreamConsumer.java

@@ -1,22 +0,0 @@
-package org.dromara.stream.mq.consumer;
-
-import org.dromara.stream.mq.TestMessaging;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.context.annotation.Bean;
-import org.springframework.stereotype.Component;
-
-import java.util.function.Consumer;
-
-@Slf4j
-@Component
-public class LogStreamConsumer {
-
-    @Bean
-    Consumer<TestMessaging> log() {
-        log.info("初始化订阅");
-        return msg -> {
-            log.info("通过stream消费到消息 => {}", msg.toString());
-        };
-    }
-
-}

ruoyi-example/ruoyi-stream-mq/src/main/java/org/dromara/stream/mq/consumer/TestStreamConsumer.java

@@ -1,22 +0,0 @@
-package org.dromara.stream.mq.consumer;
-
-import org.dromara.stream.mq.TestMessaging;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.context.annotation.Bean;
-import org.springframework.stereotype.Component;
-
-import java.util.function.Consumer;
-
-@Slf4j
-@Component
-public class TestStreamConsumer {
-
-    @Bean
-    Consumer<TestMessaging> demo() {
-        log.info("初始化订阅");
-        return msg -> {
-            log.info("通过stream消费到消息 => {}", msg.toString());
-        };
-    }
-
-}

ruoyi-example/ruoyi-stream-mq/src/main/java/org/dromara/stream/mq/producer/DelayProducer.java

@@ -1,27 +0,0 @@
-package org.dromara.stream.mq.producer;
-
-import org.dromara.stream.mq.TestMessaging;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.stream.function.StreamBridge;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.support.MessageBuilder;
-import org.springframework.stereotype.Component;
-
-import java.util.UUID;
-
-@Component
-public class DelayProducer {
-
-    @Autowired
-    private StreamBridge streamBridge;
-
-    public void sendMsg(String msg, Long delay) {
-        // 构建消息对象
-        TestMessaging testMessaging = new TestMessaging()
-            .setMsgId(UUID.randomUUID().toString())
-            .setMsgText(msg);
-        Message<TestMessaging> message = MessageBuilder.withPayload(testMessaging)
-            .setHeader("x-delay", delay).build();
-        streamBridge.send("delay-out-0", message);
-    }
-}

ruoyi-example/ruoyi-stream-mq/src/main/java/org/dromara/stream/mq/producer/LogStreamProducer.java

@@ -1,24 +0,0 @@
-package org.dromara.stream.mq.producer;
-
-import org.dromara.stream.mq.TestMessaging;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.stream.function.StreamBridge;
-import org.springframework.messaging.support.MessageBuilder;
-import org.springframework.stereotype.Component;
-
-import java.util.UUID;
-
-@Component
-public class LogStreamProducer {
-
-    @Autowired
-    private StreamBridge streamBridge;
-
-    public void streamLogMsg(String msg) {
-        // 构建消息对象
-        TestMessaging testMessaging = new TestMessaging()
-            .setMsgId(UUID.randomUUID().toString())
-            .setMsgText(msg);
-        streamBridge.send("log-out-0", MessageBuilder.withPayload(testMessaging).build());
-    }
-}

ruoyi-example/ruoyi-stream-mq/src/main/java/org/dromara/stream/mq/producer/TestStreamProducer.java

@@ -1,24 +0,0 @@
-package org.dromara.stream.mq.producer;
-
-import org.dromara.stream.mq.TestMessaging;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.stream.function.StreamBridge;
-import org.springframework.messaging.support.MessageBuilder;
-import org.springframework.stereotype.Component;
-
-import java.util.UUID;
-
-@Component
-public class TestStreamProducer {
-
-    @Autowired
-    private StreamBridge streamBridge;
-
-    public void streamTestMsg(String msg) {
-        // 构建消息对象
-        TestMessaging testMessaging = new TestMessaging()
-            .setMsgId(UUID.randomUUID().toString())
-            .setMsgText(msg);
-        streamBridge.send("demo-out-0", MessageBuilder.withPayload(testMessaging).build());
-    }
-}

ruoyi-example/ruoyi-stream-mq/src/main/resources/application.yml

@@ -1,109 +0,0 @@
-server:
-  port: 9402
-
-# Spring
-spring:
-  application:
-    # 应用名称
-    name: ruoyi-stream-mq
-  profiles:
-    # 环境配置
-    active: @profiles.active@
-  cloud:
-    stream:
-      function:
-        # 重点配置 与 binding 名与消费者对应
-        definition: delay;demo;log
-
---- # rabbitmq 配置
-spring:
-  rabbitmq:
-    host: localhost
-    port: 5672
-    username: root
-    password: root
-  cloud:
-    stream:
-      rabbit:
-        bindings:
-          delay-in-0:
-            consumer:
-              delayedExchange: true
-          delay-out-0:
-            producer:
-              delayedExchange: true
-      bindings:
-        delay-in-0:
-          destination: delay.exchange.cloud
-          content-type: application/json
-          group: delay-group
-          binder: rabbit
-        delay-out-0:
-          destination: delay.exchange.cloud
-          content-type: application/json
-          group: delay-group
-          binder: rabbit
-
---- # rocketmq 配置
-spring:
-  cloud:
-    stream:
-      rocketmq:
-        binder:
-          # rocketmq 地址
-          name-server: localhost:9876
-        bindings:
-          demo-out-0:
-            producer:
-              # 必须得写
-              group: default
-      bindings:
-        demo-out-0:
-          content-type: application/json
-          destination: stream-test-topic
-          group: test-group
-          binder: rocketmq
-        demo-in-0:
-          content-type: application/json
-          destination: stream-test-topic
-          group: test-group
-          binder: rocketmq
-
---- # kafka 配置
-spring:
-  cloud:
-    stream:
-      kafka:
-        binder:
-          brokers: localhost:9092
-      bindings:
-        log-out-0:
-          destination: stream-log-topic
-          contentType: application/json
-          group: log_group
-          binder: kafka
-        log-in-0:
-          destination: stream-log-topic
-          contentType: application/json
-          group: log_group
-          binder: kafka
-
---- # nacos 配置
-spring:
-  cloud:
-    nacos:
-      # nacos 服务地址
-      server-addr: @nacos.server@
-      username: @nacos.username@
-      password: @nacos.password@
-      discovery:
-        # 注册组
-        group: @nacos.discovery.group@
-        namespace: ${spring.profiles.active}
-      config:
-        # 配置组
-        group: @nacos.config.group@
-        namespace: ${spring.profiles.active}
-  config:
-    import:
-      - optional:nacos:application-common.yml

ruoyi-example/ruoyi-stream-mq/src/main/resources/banner.txt

@@ -1,10 +0,0 @@
-Spring Boot Version: ${spring-boot.version}
-Spring Application Name: ${spring.application.name}
-                        _            _
-                       (_)          | |
- _ __ _   _  ___  _   _ _ ______ ___| |_ _ __ ___  __ _ _ __ ___ ______ _ __ ___   __ _
-| '__| | | |/ _ \| | | | |______/ __| __| '__/ _ \/ _` | '_ ` _ \______| '_ ` _ \ / _` |
-| |  | |_| | (_) | |_| | |      \__ \ |_| | |  __/ (_| | | | | | |     | | | | | | (_| |
-|_|   \__,_|\___/ \__, |_|      |___/\__|_|  \___|\__,_|_| |_| |_|     |_| |_| |_|\__, |
-                   __/ |                                                             | |
-                  |___/                                                              |_|

ruoyi-example/ruoyi-test-mq/README.md

@@ -0,0 +1,102 @@
+# ruoyi-test-mq
+
+## 模块说明
+
+该模块基于需求【[将spring-cloud-stream改为普通spring的mq依赖用法】下修改编写；
+
+原模块缺点：功能复杂学习成本高 大部分用户用不明白 功能封闭 特性无法使用 项目中基本不会有切换mq的事情发生；
+
+现模块：集成基础的rabbit、rocketmq、kafka等主流的中间件，功能包含
+
+1. rabbit: 普通消息、延迟队列
+2. rocket：普通消息、事务消息
+3. kafka：普通消息
+
+后续可实现的：
+
+1. kafka stream流的使用
+2. rocket 顺序、异步、延时等
+
+## 项目目录
+
+```xml
+├─src
+│  └─main
+│      ├─java
+│      │  └─org
+│      │      └─dromara
+│      │          └─stream
+│      │              │  RuoYiTestMqApplication.java
+│      │              │  
+│      │              ├─config
+│      │              │      RabbitConfig.java 普通消息配置类
+│      │              │      RabbitTtlQueueConfig.java  延迟队列配置类
+│      │              │      
+│      │              ├─controller  测试类
+│      │              │      PushMessageController.java
+│      │              │      
+│      │              └─mq
+│      │                  ├─consumer
+│      │                  │  ├─kafkaMq
+│      │                  │  │      KafkaNormalConsumer.java
+│      │                  │  ├─rabbit
+│      │                  │  │      ConsumerListener.java
+│      │                  │  └─rocketmq
+│      │                  │          NormalRocketConsumer.java
+│      │                  │          TransactionRocketConsumer.java
+│      │                  ├─listener
+│      │                  │      TranscationRocketListener.java
+│      │                  └─producer
+│      │                      ├─kafkaMq
+│      │                      │      KafkaNormalProducer.java
+│      │                      ├─rabbitMq
+│      │                      │      DelayRabbitProducer.java
+│      │                      │      NormalRabbitProducer.java
+│      │                      └─rocketMq
+│      │  	                         NormalRocketProducer.java
+│      │  	                         TransactionRocketProducer.java
+│      │                                  
+│      └─resources
+│              application.yml  IP:Host根据实际情况替换
+│              logback-plus.xml
+```
+
+## 使用方式
+
+rocketmq：
+
+**注意：需要进入到rockerMQ文件路径中执行**
+
+创建普通消息的topic
+
+```sh
+sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=NORMAL
+```
+
+```shell
+bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster
+```
+
+创建事务消息的topic
+
+```sh
+sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=TRANSACTION
+```
+
+```shell
+bin/mqadmin updatetopic -n localhost:9876 -t transaction_topic -c DefaultCluster -a +message.type=TRANSACTION
+```
+
+kafka:
+
+```shell
+kafka-topics.sh --create --topic <topic_name> --bootstrap-server <broker_list> --partitions <num_partitions> --replication-factor <replication_factor>
+```
+
+```shell
+kafka-topics.sh --create --topic my_topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
+```
+
+## 验证方式
+
+可通过`PushMessageController`实现`Restful`进行测试;

ruoyi-example/ruoyi-stream-mq/pom.xml → ruoyi-example/ruoyi-test-mq/pom.xml

@@ -9,10 +9,10 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>ruoyi-stream-mq</artifactId>
+    <artifactId>ruoyi-test-mq</artifactId>
 
     <description>
-        ruoyi-stream-mq SpringCloud-Stream-MQ 案例项目
+        ruoyi-test-mq 案例项目
     </description>
 
     <dependencies>
@@ -21,22 +21,20 @@
             <groupId>org.dromara</groupId>
             <artifactId>ruoyi-common-nacos</artifactId>
         </dependency>
-
         <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
         </dependency>
-
         <dependency>
-            <groupId>com.alibaba.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-spring-boot-starter</artifactId>
         </dependency>
-
         <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
         </dependency>
 
+
         <dependency>
             <groupId>org.dromara</groupId>
             <artifactId>ruoyi-common-sentinel</artifactId>

ruoyi-example/ruoyi-stream-mq/src/main/java/org/dromara/stream/RuoYiStreamMqApplication.java → ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/RuoYiTestMqApplication.java

@@ -5,15 +5,14 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.metrics.buffering.BufferingApplicationStartup;
 
 /**
- * SpringCloud-Stream-MQ 案例项目
- *
+ * SpringBoot-MQ 案例项目
  * @author Lion Li
  */
 @SpringBootApplication
-public class RuoYiStreamMqApplication {
+public class RuoYiTestMqApplication {
 
     public static void main(String[] args) {
-        SpringApplication application = new SpringApplication(RuoYiStreamMqApplication.class);
+        SpringApplication application = new SpringApplication(RuoYiTestMqApplication.class);
         application.setApplicationStartup(new BufferingApplicationStartup(2048));
         application.run(args);
         System.out.println("(♥◠‿◠)ﾉﾞ  MQ案例模块启动成功   ლ(´ڡ`ლ)ﾞ  ");

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/config/RabbitConfig.java

@@ -0,0 +1,54 @@
+package org.dromara.stream.config;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author xbhog
+ */
+@Configuration
+public class RabbitConfig {
+
+    public static final String EXCHANGE_NAME = "demo-exchange";
+    public static final String QUEUE_NAME = "demo-queue";
+    public static final String ROUTING_KEY = "demo.routing.key";
+    /**
+     * 创建交换机
+     * ExchangeBuilder有四种交换机模式
+	 * Direct Exchange：直连交换机，根据Routing Key(路由键)进行投递到不同队列。
+     * Fanout Exchange：扇形交换机，采用广播模式，根据绑定的交换机，路由到与之对应的所有队列。
+     * Topic Exchange：主题交换机，对路由键进行模式匹配后进行投递，符号#表示一个或多个词，*表示一个词。
+     * Header Exchange：头交换机，不处理路由键。而是根据发送的消息内容中的headers属性进行匹配。
+     * durable 交换器是否持久化（false 不持久化，true 持久化）
+     **/
+    @Bean
+    public TopicExchange exchange() {
+        return new TopicExchange(EXCHANGE_NAME);
+    }
+	/**
+	 * 创建队列
+	 * durable 队列是否持久化 队列调用此方法就是持久化 可查看方法的源码
+	 * deliveryMode 消息是否持久化（1 不持久化，2 持久化）
+	 **/
+    @Bean
+    public Queue queue() {
+        return new Queue(QUEUE_NAME, false);
+    }
+	/**
+	* 绑定交换机和队列
+	* bing 方法参数可以是队列和交换机
+	* to 方法参数必须是交换机
+	* with 方法参数是路由Key 这里是以rabbit.开头
+	* noargs 就是不要参数的意思
+	* 这个方法的意思是把rabbit开头的消息 和 上面的队列 和 上面的交换机绑定
+	**/
+    @Bean
+    public Binding binding(Queue queue, TopicExchange exchange) {
+        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
+    }
+
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/config/RabbitTtlQueueConfig.java

@@ -0,0 +1,69 @@
+package org.dromara.stream.config;
+
+import org.springframework.amqp.core.*;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+
+/**
+ * RabbitTTL队列
+ * @author xbhog
+ */
+@Configuration
+public class RabbitTtlQueueConfig {
+
+    // 延迟队列名称
+    public static final String DELAY_QUEUE_NAME = "delay-queue";
+    // 延迟交换机名称
+    public static final String DELAY_EXCHANGE_NAME = "delay-exchange";
+    // 延迟路由键名称
+    public static final String DELAY_ROUTING_KEY = "delay.routing.key";
+
+    // 死信交换机名称
+    public static final String DEAD_LETTER_EXCHANGE = "dlx-exchange";
+    // 死信队列名称
+    public static final String DEAD_LETTER_QUEUE = "dlx-queue";
+    // 死信路由键名称
+    public static final String DEAD_LETTER_ROUTING_KEY = "dlx.routing.key";
+    // 延迟消息的默认 TTL（毫秒）
+    @Value("${rabbitmq.delay.ttl:5000}")
+    private long messageTTL;
+
+    // 声明延迟队列
+    @Bean
+    public Queue delayQueue() {
+        return QueueBuilder.durable(DELAY_QUEUE_NAME)
+            .withArgument("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE)
+            .withArgument("x-dead-letter-routing-key", DEAD_LETTER_ROUTING_KEY)
+            .withArgument("x-message-ttl", messageTTL)
+            .build();
+    }
+    // 声明延迟交换机
+    @Bean
+    public TopicExchange delayExchange() {
+        return new TopicExchange(DELAY_EXCHANGE_NAME);
+    }
+    // 将延迟队列绑定到延迟交换机
+    @Bean
+    public Binding delayBinding(Queue delayQueue, TopicExchange delayExchange) {
+        return BindingBuilder.bind(delayQueue).to(delayExchange).with(DELAY_ROUTING_KEY);
+    }
+    // 声明死信队列
+    @Bean
+    public Queue deadLetterQueue() {
+        return new Queue(DEAD_LETTER_QUEUE);
+    }
+    // 声明死信交换机
+    @Bean
+    public TopicExchange deadLetterExchange() {
+        return new TopicExchange(DEAD_LETTER_EXCHANGE);
+    }
+
+    // 将死信队列绑定到死信交换机
+    @Bean
+    public Binding deadLetterBinding(Queue deadLetterQueue, TopicExchange deadLetterExchange) {
+        return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with(DEAD_LETTER_ROUTING_KEY);
+    }
+}
+

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/controller/PushMessageController.java

@@ -0,0 +1,72 @@
+package org.dromara.stream.controller;
+
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.stream.mq.producer.kafkaMq.KafkaNormalProducer;
+import org.dromara.stream.mq.producer.rabbitMq.DelayRabbitProducer;
+import org.dromara.stream.mq.producer.rabbitMq.NormalRabbitProducer;
+import org.dromara.stream.mq.producer.rocketMq.NormalRocketProducer;
+import org.dromara.stream.mq.producer.rocketMq.TransactionRocketProducer;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author xbhog
+ */
+@Slf4j
+@RestController
+@RequestMapping("push/message")
+public class PushMessageController {
+
+    @Resource
+    private NormalRabbitProducer normalRabbitProducer;
+
+    @Resource
+    private DelayRabbitProducer delayRabbitProducer;
+
+    @Resource
+    private NormalRocketProducer normalRocketProducer;
+
+    @Resource
+    private TransactionRocketProducer transactionRocketProducer;
+
+    @Resource
+    private KafkaNormalProducer normalKafkaProducer;
+
+    /**
+     * rabbit普通消息的处理
+     */
+    @GetMapping("/rabbitMsg/sendNormal")
+    public void sendMq() {
+        normalRabbitProducer.sendMq("hello normal RabbitMsg");
+    }
+
+    /**
+     * rabbit延迟队列类型：类似生产者
+     */
+    @GetMapping("/rabbitMsg/sendDelay")
+    public void sendMessage() {
+        delayRabbitProducer.sendDelayMessage("Hello ttl RabbitMsg");
+    }
+
+    /**
+     * rockerMQ实例
+     * 需要手动创建相关的Topic和group
+     */
+    @GetMapping("/rocketMq/send")
+    public void sendRockerMq(){
+        normalRocketProducer.sendMessage();
+    }
+    @GetMapping("/rocketMq/transactionMsg")
+    public void sendRockerMqTransactionMsg(){
+        transactionRocketProducer.sendTransactionMessage();
+    }
+    /**
+     * kafkaSpringboot集成
+     */
+    @GetMapping("/kafkaMsg/send")
+    public void sendKafkaMsg(){
+        normalKafkaProducer.sendKafkaMsg();
+    }
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/consumer/kafkaMq/KafkaNormalConsumer.java

@@ -0,0 +1,24 @@
+package org.dromara.stream.mq.consumer.kafkaMq;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xbhog
+ * @date 2024/05/19 18:04
+ **/
+@Slf4j
+@Component
+public class KafkaNormalConsumer {
+
+    //默认获取最后一条消息
+    @KafkaListener(topics = "test-topic",groupId = "demo")
+    public void timiKafka(ConsumerRecord record){
+        Object key = record.key();
+        Object value = record.value();
+        log.info("【消费者】received the message key {}，value：{}",key,value);
+    }
+
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/consumer/rabbit/ConsumerListener.java

@@ -0,0 +1,36 @@
+package org.dromara.stream.mq.consumer.rabbit;
+
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.stream.config.RabbitConfig;
+import org.dromara.stream.config.RabbitTtlQueueConfig;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xbhog
+ * @date 2024年5月18日
+ */
+@Slf4j
+@Component
+public class ConsumerListener {
+
+    /**
+     * 设置监听哪一个队列 这个队列是RabbitConfig里面设置好的队列名字
+     * 普通消息
+     **/
+    @RabbitListener(queues = RabbitConfig.QUEUE_NAME)
+    public void listenQueue(Message message) {
+        log.info("【消费者】Start consuming data：{}",new String(message.getBody()));
+    }
+
+    /**
+     * 处理延迟队列的操作
+     * 该部分处理的延迟操作在消费上的时间可能与设置的TTl不同；
+     * 一般会超长；原因是消息可能并不会按时死亡；可通过延迟队列插件处理
+     */
+    @RabbitListener(queues = RabbitTtlQueueConfig.DEAD_LETTER_QUEUE)
+    public void receiveMessage(String message){
+        log.info("【消费者】Received delayed message：{}",message);
+    }
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/consumer/rocketmq/NormalRocketConsumer.java

@@ -0,0 +1,20 @@
+package org.dromara.stream.mq.consumer.rocketmq;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xbhog
+ * @date 2024/06/01 16:53
+ **/
+@Slf4j
+@Component
+@RocketMQMessageListener(topic = "TestTopic", consumerGroup = "springboot-mq-consumer-1")
+public class NormalRocketConsumer implements RocketMQListener<String> {
+    @Override
+    public void onMessage(String message) {
+        log.info("【消费者】接收消息：{}" ,message);
+    }
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/consumer/rocketmq/TransactionRocketConsumer.java

@@ -0,0 +1,21 @@
+package org.dromara.stream.mq.consumer.rocketmq;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xbhog
+ * @date 2024/06/01 16:54
+ **/
+@Slf4j
+@Component
+@RocketMQMessageListener(consumerGroup = "transaction-group", topic = "transaction_topic")
+public class TransactionRocketConsumer  implements RocketMQListener<String> {
+
+    @Override
+    public void onMessage(String message) {
+        log.info("【消费者】===>接收事务消息：{}",message);
+    }
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/listener/TranscationRocketListener.java

@@ -0,0 +1,42 @@
+package org.dromara.stream.mq.listener;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
+import org.springframework.messaging.Message;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xbhog
+ * @date 2024/06/01 17:05
+ **/
+@Slf4j
+@Component
+@RocketMQTransactionListener
+public class TranscationRocketListener implements RocketMQLocalTransactionListener {
+    @Override
+    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
+        log.info("执行本地事务");
+        String tag = String.valueOf(message.getHeaders().get("rocketmq_TAGS"));
+        if ("TAG-1".equals(tag)) {
+            //这里只讲TAGA消息提交，状态为可执行
+            log.info("【监听器】这里是校验TAG-1;提交状态:COMMIT");
+            return RocketMQLocalTransactionState.COMMIT;
+        } else if ("TAG-2".equals(tag)) {
+            log.info("【监听器】这里是校验TAG-2;提交状态:ROLLBACK");
+            return RocketMQLocalTransactionState.ROLLBACK;
+        } else if ("TAG-3".equals(tag)) {
+            log.info("【监听器】这里是校验TAG-3;提交状态:UNKNOWN");
+            return RocketMQLocalTransactionState.UNKNOWN;
+        }
+        log.info("=========【监听器】提交状态:UNKNOWN");
+        return RocketMQLocalTransactionState.UNKNOWN;
+    }
+
+    @Override
+    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
+        log.info("【监听器】检查本地交易===>{}", message);
+        return RocketMQLocalTransactionState.COMMIT;
+    }
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/producer/kafkaMq/KafkaNormalProducer.java

@@ -0,0 +1,24 @@
+package org.dromara.stream.mq.producer.kafkaMq;
+
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * @author xbhog
+ * @date 2024/05/19 18:02
+ **/
+@Slf4j
+@Component
+public class KafkaNormalProducer {
+    @Resource
+    private KafkaTemplate kafkaTemplate;
+
+    public void sendKafkaMsg(){
+        CompletableFuture send = kafkaTemplate.send("test-topic","hello", "kafkaTest");
+        send.join();
+    }
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/producer/rabbitMq/DelayRabbitProducer.java

@@ -0,0 +1,25 @@
+package org.dromara.stream.mq.producer.rabbitMq;
+
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.stream.config.RabbitTtlQueueConfig;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.web.bind.annotation.GetMapping;
+
+/**
+ * @author xbhog
+ * @date 2024/05/25 17:15
+ **/
+@Slf4j
+@Component
+public class DelayRabbitProducer {
+    @Resource
+    private RabbitTemplate rabbitTemplate;
+
+    @GetMapping("/sendDelay")
+    public void sendDelayMessage(String message) {
+        rabbitTemplate.convertAndSend(RabbitTtlQueueConfig.DELAY_EXCHANGE_NAME, RabbitTtlQueueConfig.DELAY_ROUTING_KEY, message);
+        log.info("【生产者】Delayed message send: " + message);
+    }
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/producer/rabbitMq/NormalRabbitProducer.java

@@ -0,0 +1,24 @@
+package org.dromara.stream.mq.producer.rabbitMq;
+
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.stream.config.RabbitConfig;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xbhog
+ */
+@Slf4j
+@Component
+public class NormalRabbitProducer {
+
+    @Resource
+    RabbitTemplate rabbitTemplate;
+
+
+    public void sendMq(String message) {
+        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME, RabbitConfig.ROUTING_KEY, message);
+        log.info("【生产者】Message send: " + message);
+    }
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/producer/rocketMq/NormalRocketProducer.java

@@ -0,0 +1,25 @@
+package org.dromara.stream.mq.producer.rocketMq;
+
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xbhog
+ * @date 2024/06/01 16:49
+ **/
+@Slf4j
+@Component
+public class NormalRocketProducer {
+
+    @Resource
+    private RocketMQTemplate rocketMQTemplate;
+
+    public void sendMessage(){
+        SendResult sendResult = rocketMQTemplate.syncSend("TestTopic", MessageBuilder.withPayload("hello world test").build());
+        log.info("发送普通同步消息-msg，syncSendMessage===>{}", sendResult);
+    }
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/mq/producer/rocketMq/TransactionRocketProducer.java

@@ -0,0 +1,41 @@
+package org.dromara.stream.mq.producer.rocketMq;
+
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.client.producer.TransactionSendResult;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Component;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * @author xbhog
+ * @date 2024/06/01 16:54
+ **/
+@Slf4j
+@Component
+public class TransactionRocketProducer {
+
+    @Resource
+    private RocketMQTemplate rocketMQTemplate;
+
+    public void sendTransactionMessage(){
+        List<String> tags = Arrays.asList("TAG-1", "TAG-2", "TAG-3");
+        for (int i = 0; i < 3; i++) {
+            Message<String> message = MessageBuilder.withPayload("===>事务消息-" + i).build();
+            //destination formats: `topicName:tags` message – message Message arg – ext arg
+            TransactionSendResult res = rocketMQTemplate.sendMessageInTransaction("transaction_topic:" + tags.get(i), message, i + 1);
+            if (res.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE) && res.getSendStatus().equals(SendStatus.SEND_OK)) {
+                log.info("【生产者】事物消息发送成功；成功结果：{}",res);
+            }else{
+                log.info("【生产者】事务发送失败：失败原因：{}",res);
+            }
+        }
+    }
+
+}

ruoyi-example/ruoyi-test-mq/src/main/resources/application.yml

@@ -0,0 +1,44 @@
+server:
+  port: 9402
+
+# Spring
+spring:
+  application:
+    # 应用名称
+    name: ruoyi-test-mq
+  profiles:
+    # 环境配置
+    active: @profiles.active@
+  #MQ配置
+  rabbitmq:
+    host: 192.168.1.13
+    port: 5672
+    username: mq
+    password: mq
+    publisher-returns: true
+    publisher-confirm-type: correlated
+  kafka:
+    bootstrap-servers: 192.168.1.13:9092
+    producer:
+      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
+rocketmq:
+    name-server: 192.168.1.13:9876
+    producer:
+      group: dist-test # 生产者组
+--- # nacos 配置
+spring:
+  cloud:
+    nacos:
+      # nacos 服务地址
+      server-addr: @nacos.server@
+      discovery:
+        # 注册组
+        group: @nacos.discovery.group@
+        namespace: ${spring.profiles.active}
+      config:
+        # 配置组
+        group: @nacos.config.group@
+        namespace: ${spring.profiles.active}
+  config:
+    import:
+      - optional:nacos:application-common.yml

ruoyi-example/ruoyi-stream-mq/src/main/resources/logback-plus.xml → ruoyi-example/ruoyi-test-mq/src/main/resources/logback-plus.xml

@@ -19,7 +19,8 @@
     <!-- 开启 skywalking 日志收集 -->
     <include resource="logback-skylog.xml" />
 
-	<!--系统操作日志-->
+
+    <!--系统操作日志-->
     <root level="info">
         <appender-ref ref="console" />
     </root>