MQ 示例

soc-modules/soc-modules-api/pom.xml

@@ -14,6 +14,7 @@
         soc-modules-api接口服务模块
     </description>
 
+
     <dependencies>
         <!-- SpringCloud Alibaba Nacos -->
         <dependency>
@@ -39,20 +40,25 @@
             <artifactId>spring-boot-starter-actuator</artifactId>
         </dependency>
 
-        <!-- Swagger UI -->
         <dependency>
-            <groupId>io.springfox</groupId>
-            <artifactId>springfox-swagger-ui</artifactId>
-            <version>${swagger.fox.version}</version>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
+            <version>3.0.1.RELEASE</version>
         </dependency>
 
+        <!-- Swagger UI -->
+        <!--      <dependency>
+                  <groupId>io.springfox</groupId>
+                  <artifactId>springfox-swagger-ui</artifactId>
+                  <version>${swagger.fox.version}</version>
+              </dependency>
+      -->
         <!-- Mysql Connector -->
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
         </dependency>
 
-
         <!--soc datascope 数据权限-->
         <dependency>
             <groupId>com.xunmei</groupId>
@@ -78,6 +84,50 @@
             <version>${soc.version}</version>
         </dependency>
 
+        <!-- freemarker -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-freemarker</artifactId>
+            <version>${freemarker.version}</version>
+        </dependency>
+
+        <!--代码生成器包-->
+        <dependency>
+            <groupId>com.baomidou</groupId>
+            <artifactId>mybatis-plus-generator</artifactId>
+            <version>${generator.version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>mybatis-plus-extension</artifactId>
+                    <groupId>com.baomidou</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <!--lombok-->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+
     </dependencies>
 
-</project>
+    <build>
+        <finalName>${project.artifactId}</finalName>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+
+</project>

soc-modules/soc-modules-api/src/main/java/com/xunmei/api/SocApiApplication.java

@@ -1,7 +1,21 @@
 package com.xunmei.api;
 
+
+import com.xunmei.common.security.annotation.EnableCustomConfig;
+import com.xunmei.common.security.annotation.EnableSocFeignClients;
+import com.xunmei.common.swagger.annotation.EnableCustomSwagger2;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableAsync;
+
+@EnableAsync
+@EnableCustomConfig
+@EnableCustomSwagger2
+@EnableSocFeignClients
+@SpringBootApplication
 public class SocApiApplication {
     public static void main(String[] args) {
-        System.out.println("Hello world!");
+        SpringApplication.run(SocApiApplication.class, args);
+        System.out.println("(♥◠‿◠)ﾉﾞ  系统模块接口启动成功   ლ(´ڡ`ლ)ﾞ  ");
     }
-}
+}

soc-modules/soc-modules-api/src/main/java/com/xunmei/api/controller/MqController.java

@@ -0,0 +1,58 @@
+package com.xunmei.api.controller;
+
+import com.alibaba.fastjson2.JSON;
+import com.xunmei.api.mq.producer.RabbitMqProducer;
+import com.xunmei.common.core.domain.R;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.*;
+
+import java.time.LocalDateTime;
+
+/**
+ * @Description: RabbitmqController
+ */
+@Slf4j
+@RestController
+@AllArgsConstructor
+@RequestMapping("/mq")
+public class MqController {
+
+    //消息发送者
+    private final RabbitMqProducer rabbitMqProducer;
+
+    /**
+     * 发送普通消息Rabbitmq
+     * bindingName 绑定队列名称
+     * @param msg 消息内容
+     */
+    @GetMapping("/sendMessage/{msg}/{bindingName}")
+    public R<Void> sendMessage(@PathVariable("msg") String msg, @PathVariable("bindingName") String bindingName) {
+        log.info(bindingName + "发送消息: " + msg);
+        rabbitMqProducer.sendMsg(msg, bindingName);
+        return R.ok();
+    }
+
+    /**
+     * 发送延迟消息
+     *
+     * @param message  消息实体
+     * @return
+     */
+    @PostMapping("/sendDelayedMessage")
+    public R<Void> sendDelayedMessage(@RequestBody Message message) {
+        log.info("发送延时消息: " + LocalDateTime.now() + "  " + message);
+        rabbitMqProducer.sendDelayMsg(JSON.toJSONString(message), message.getBindingName(), message.getSeconds());// 延迟时间（秒）
+        return R.ok();
+    }
+
+}
+
+@Data
+class  Message{
+    private String bindingName;
+
+    private Integer seconds;
+}
+

soc-modules/soc-modules-api/src/main/java/com/xunmei/api/mq/Messaging.java

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

soc-modules/soc-modules-api/src/main/java/com/xunmei/api/mq/consumer/RabbitMqConsumer.java

@@ -0,0 +1,78 @@
+package com.xunmei.api.mq.consumer;
+
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
+import com.rabbitmq.client.Channel;
+import com.xunmei.api.mq.Messaging;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.HttpHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.stereotype.Component;
+import org.springframework.util.MultiValueMap;
+
+import java.time.LocalDateTime;
+import java.util.function.Consumer;
+
+
+/**
+ * RabbitMq消息消费者
+ */
+@Component
+@Slf4j
+public class RabbitMqConsumer {
+
+    /**
+     * mq接收ackMessage消息/手动ack确认
+     * @methodName 配置文件对应
+     **/
+    @Bean
+    Consumer<Message<Messaging>> ackMessage() {
+        log.info("ackMessage-初始化订阅");
+        return obj -> {
+            Channel channel = obj.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
+            Long deliveryTag = obj.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class);
+            try {
+                log.info("ackMessage-消息接收成功：" + obj.getPayload());
+                //业务逻辑处理
+                //ack确认
+                channel.basicAck(deliveryTag, false);
+            } catch (Exception e) {
+                //重新回队列-true则重新入队列，否则丢弃或者进入死信队列。
+//                    channel.basicReject(deliveryTag, true);
+                log.error(e.getMessage());
+            }
+
+        };
+    }
+
+    /**
+     * mq接收normal消息
+     **/
+    @Bean
+    Consumer<Messaging> normal() {
+        log.info("normal-初始化订阅");
+        return obj -> {
+            log.info("normal-消息接收成功：" + obj);
+            //业务逻辑处理
+        };
+    }
+
+
+    /**
+     * mq接收延时消息
+     * Messaging 发送实体消息接收实体消息
+     **/
+    @Bean
+    Consumer<Message<Messaging>> delay() {
+        log.info("delay-初始化订阅");
+        return obj -> {
+            Messaging payload = obj.getPayload();
+            log.info("delay-消息接收成功：" + LocalDateTime.now() + "  " + payload);
+            //业务逻辑处理
+        };
+    }
+}
+

soc-modules/soc-modules-api/src/main/java/com/xunmei/api/mq/producer/RabbitMqProducer.java

@@ -0,0 +1,47 @@
+package com.xunmei.api.mq.producer;
+
+import com.xunmei.api.mq.Messaging;
+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;
+
+/**
+ * RabbitMq消息生产者
+ */
+@Component
+public class RabbitMqProducer {
+
+    @Autowired
+    private StreamBridge streamBridge;
+
+    /**
+     * @Description RabbitMq消息生产者
+     * @Param msg 消息内容
+     * @Param bindingName  exchange绑定queue名称
+     **/
+    public void sendMsg(String msg, String bindingName) {
+        // 构建消息对象
+        Messaging messaging = new Messaging().setMsgId(UUID.randomUUID().toString()).setMsgText(msg);
+        Message<Messaging> message = MessageBuilder.withPayload(messaging).build();
+        streamBridge.send(bindingName, message);
+    }
+
+    /**
+     * 发送延迟消息
+     *
+     * @param msg
+     * @param bindingName
+     * @param seconds
+     */
+    public void sendDelayMsg(String msg, String bindingName, Integer seconds) {
+        // 构建消息对象
+        Messaging messaging = new Messaging().setMsgId(UUID.randomUUID().toString()).setMsgText(msg);
+        Message<Messaging> message = MessageBuilder.withPayload(messaging).setHeader("x-delay", seconds * 1000).build();
+        streamBridge.send(bindingName, message);
+    }
+}
+

soc-modules/soc-modules-api/src/main/resources/bootstrap.yml

@@ -3,7 +3,7 @@ server:
   port: 9003
 
 # Spring
-spring: 
+spring:
   application:
     # 应用名称
     name: soc-api
@@ -13,11 +13,11 @@ spring:
   cloud:
     nacos:
       discovery:
-        namespace: 34306a91-1bb7-45ce-b80d-4092dd08ea64
+        namespace: 7e29a19d-6b00-4918-9a79-8f80d15d2002
         # 服务注册地址
         server-addr: 10.87.10.54:8848
       config:
-        namespace: 34306a91-1bb7-45ce-b80d-4092dd08ea64
+        namespace: 7e29a19d-6b00-4918-9a79-8f80d15d2002
         # 配置中心地址
         server-addr: 10.87.10.54:8848
         # 配置文件格式
@@ -27,4 +27,4 @@ spring:
           - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
 logging:
   file:
-    name: logs/${spring.application.name}/info.log
+    name: logs/${spring.application.name}/info.log