From 965b9d56d1f033b63b032a5767f1eeb4f5d6fbbb Mon Sep 17 00:00:00 2001 From: Leo Date: Sun, 1 Dec 2024 02:20:20 +0800 Subject: [PATCH] 1 --- pom.xml | 6 ++- .../business/mq/MessageConsumerService.java | 53 +++++++++++++++++++ .../business/mq/MessageProducerService.java | 22 ++++++++ .../java/cn/van/business/util/WXUtil.java | 6 ++- src/main/resources/application-dev.yml | 10 ++++ 5 files changed, 95 insertions(+), 2 deletions(-) create mode 100644 src/main/java/cn/van/business/mq/MessageConsumerService.java create mode 100644 src/main/java/cn/van/business/mq/MessageProducerService.java diff --git a/pom.xml b/pom.xml index 54d721b..4418ad6 100644 --- a/pom.xml +++ b/pom.xml @@ -50,7 +50,11 @@ mysql-connector-java runtime - + + org.apache.rocketmq + rocketmq-spring-boot-starter + 2.2.0 + org.springframework.boot diff --git a/src/main/java/cn/van/business/mq/MessageConsumerService.java b/src/main/java/cn/van/business/mq/MessageConsumerService.java new file mode 100644 index 0000000..3116f9e --- /dev/null +++ b/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 { + + 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 可以捕获到这个异常 + } + + } +} diff --git a/src/main/java/cn/van/business/mq/MessageProducerService.java b/src/main/java/cn/van/business/mq/MessageProducerService.java new file mode 100644 index 0000000..8695cbc --- /dev/null +++ b/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); + } +} diff --git a/src/main/java/cn/van/business/util/WXUtil.java b/src/main/java/cn/van/business/util/WXUtil.java index 8ba08e4..373f7fb 100644 --- a/src/main/java/cn/van/business/util/WXUtil.java +++ b/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)) { diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index bc5a54a..3dbb781 100644 --- a/src/main/resources/application-dev.yml +++ b/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 # 批量消费最大消息数