1

pom.xml

@@ -50,7 +50,11 @@
             <artifactId>mysql-connector-java</artifactId>
             <scope>runtime</scope>
         </dependency>
-
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-spring-boot-starter</artifactId>
+            <version>2.2.0</version>
+        </dependency>
         <!-- Spring Boot Starter Test，测试依赖 -->
         <dependency>
             <groupId>org.springframework.boot</groupId>

src/main/java/cn/van/business/mq/MessageConsumerService.java

@@ -0,0 +1,53 @@
+package cn.van.business.mq;
+
+import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.http.HttpRequest;
+import cn.hutool.json.JSONUtil;
+import com.alibaba.fastjson2.JSONObject;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import static cn.van.business.util.WXUtil.WX_BASE_URL;
+
+/**
+ * @author Leo
+ * @version 1.0
+ * @create 2024/12/1 上午2:06
+ * @description：
+ */
+@Service
+@RocketMQMessageListener(topic = "wx-message", consumerGroup = "${rocketmq.consumer.group}")
+public class MessageConsumerService implements RocketMQListener<JSONObject> {
+
+    private static final Logger logger = LoggerFactory.getLogger(MessageConsumerService.class);
+
+    @Override
+    public void onMessage(JSONObject jsonObject) {
+        // 处理消息
+        try {
+            String responseStr = HttpRequest.post(WX_BASE_URL)
+                    .body(JSONUtil.toJsonStr(jsonObject))
+                    .execute()
+                    .body();
+
+            if (ObjectUtil.isNotEmpty(responseStr)) {
+                JSONObject response = JSONObject.parseObject(responseStr);
+                if (response.getInteger("code") != 200) {
+                    // TODO: 如果需要处理错误，您可以在这里添加逻辑
+                    throw new RuntimeException("消息发送失败: " + responseStr);
+                }
+                logger.info("消息成功发送并得到响应：{}", response);
+            } else {
+                throw new RuntimeException("消息发送失败，没有收到响应");
+
+            }
+        } catch (Exception e) {
+            logger.error("处理消息时发生错误", e);
+            throw e; // 重抛异常使得 RocketMQ 可以捕获到这个异常
+        }
+
+    }
+}

src/main/java/cn/van/business/mq/MessageProducerService.java

@@ -0,0 +1,22 @@
+package cn.van.business.mq;
+
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author Leo
+ * @version 1.0
+ * @create 2024/12/1 上午2:06
+ * @description：
+ */
+@Service
+public class MessageProducerService {
+
+    @Autowired
+    private RocketMQTemplate rocketMQTemplate;
+
+    public void sendMessage(String topic, String message) {
+        this.rocketMQTemplate.convertAndSend(topic, message);
+    }
+}

src/main/java/cn/van/business/util/WXUtil.java

@@ -8,6 +8,7 @@ import com.alibaba.fastjson2.JSONObject;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -61,7 +62,8 @@ public class WXUtil {
      */
     public static String WX_BASE_URL;
     private Environment env;
-
+    @Autowired
+    private RocketMQTemplate rocketMQTemplate;
     @Autowired
     public WXUtil(Environment env) {
         this.env = env;
@@ -137,6 +139,8 @@ public class WXUtil {
             //System.out.println(JSON.toJSONString(jsonObject));
             //wxReqDate.setData(jsonObject);
             if (Util.isNotEmpty(wxid)) {
+                // 把消息发送到RocketMQ，使用'wx-message'作为topic，jsonObject作为消息内容。
+                rocketMQTemplate.convertAndSend("wx-message", jsonObject);
                 String responseStr = HttpRequest.post(WX_BASE_URL)
                         .body(JSON.toJSONString(jsonObject)).execute().body();
                 if (ObjectUtil.isNotEmpty(responseStr)) {

src/main/resources/application-dev.yml

@@ -56,3 +56,13 @@ logging:
 config:
     WX_BASE_URL: http://192.168.8.208:7777/qianxun/httpapi?wxid=wxid_kr145nk7l0an31
     QL_BASE_URL: http://134.175.126.60:35700
+rocketmq:
+    name-server: 192.168.8.9:9876  # RocketMQ Name Server 地址
+    producer:
+        group: wx_producer  # 生产者组名
+        send-msg-timeout: 3000  # 发送消息超时时间
+    consumer:
+        group: wx_consumer  # 消费者组名
+        consume-thread-min: 20  # 消费线程池最小线程数
+        consume-thread-max: 64  # 消费线程池最大线程数
+        consume-message-batch-max-size: 20  # 批量消费最大消息数