Merge remote-tracking branch 'origin/master'

# Conflicts:
#	src/main/java/cn/van/business/mq/MessageConsumerService.java
This commit is contained in:
Leo
2025-01-24 13:03:43 +08:00
7 changed files with 83 additions and 22 deletions

View File

@@ -1,7 +1,7 @@
package cn.van.business.controller.jd; package cn.van.business.controller.jd;
import cn.van.business.mq.MessageProducerService; import cn.van.business.mq.MessageProducerService;
import cn.van.business.util.JDUtils; import cn.van.business.util.JDUtil;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.ResponseBody;
@@ -21,7 +21,7 @@ public class OrderController {
public static String TOKEN = "cc0313"; public static String TOKEN = "cc0313";
@Resource @Resource
private JDUtils jdUtils; private JDUtil jdUtils;
@Resource @Resource
private MessageProducerService messageProducerService; private MessageProducerService messageProducerService;

View File

@@ -3,17 +3,20 @@ package cn.van.business.mq;
import cn.hutool.core.util.ObjectUtil; import cn.hutool.core.util.ObjectUtil;
import cn.hutool.http.HttpRequest; import cn.hutool.http.HttpRequest;
import cn.hutool.json.JSONUtil; import cn.hutool.json.JSONUtil;
import cn.van.business.util.WxtsUtil;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.google.common.util.concurrent.RateLimiter; import com.google.common.util.concurrent.RateLimiter;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener; import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.util.Base64; import java.util.Base64;
import static cn.hutool.core.thread.ThreadUtil.sleep;
import static cn.van.business.util.WXUtil.WX_BASE_URL; import static cn.van.business.util.WXUtil.WX_BASE_URL;
/** /**
@@ -28,32 +31,41 @@ public class MessageConsumerService implements RocketMQListener<JSONObject> {
private static final Logger logger = LoggerFactory.getLogger(MessageConsumerService.class); private static final Logger logger = LoggerFactory.getLogger(MessageConsumerService.class);
private final WxtsUtil wxtsUtil;
// create a rate limiter of 1 qps // create a rate limiter of 1 qps
RateLimiter rateLimiter = RateLimiter.create(0.5); RateLimiter rateLimiter = RateLimiter.create(0.5);
@Autowired
public MessageConsumerService(WxtsUtil wxtsUtil) {
this.wxtsUtil = wxtsUtil;
}
@Override @Override
public void onMessage(JSONObject jsonObject) { public void onMessage(JSONObject jsonObject) {
// 处理消息 // 处理消息
try { try {
rateLimiter.acquire(); // 请求许可。如果超过速率,则此方法会阻塞 rateLimiter.acquire(); // 请求许可。如果超过速率,则此方法会阻塞
//logger.info("消费到消息:{}", jsonObject);
String body = jsonObject.getString("body"); String body = jsonObject.getString("body");
byte[] decodedBody = Base64.getDecoder().decode(body); byte[] decodedBody = Base64.getDecoder().decode(body);
String decodedBodyStr = new String(decodedBody, StandardCharsets.UTF_8); String decodedBodyStr = new String(decodedBody, StandardCharsets.UTF_8);
JSONObject decodedBodyJson = JSONObject.parseObject(decodedBodyStr); JSONObject decodedBodyJson = JSONObject.parseObject(decodedBodyStr);
String jsonStr = JSONUtil.toJsonStr(decodedBodyJson);
String responseStr = HttpRequest.post(WX_BASE_URL) String responseStr = HttpRequest.post(WX_BASE_URL)
.body(JSONUtil.toJsonStr(decodedBodyJson)) .body(jsonStr)
.execute() .execute()
.body(); .body();
logger.info("消费消息:{}", jsonStr);
if (ObjectUtil.isNotEmpty(responseStr)) { if (ObjectUtil.isNotEmpty(responseStr)) {
JSONObject response = JSONObject.parseObject(responseStr); JSONObject response = JSONObject.parseObject(responseStr);
logger.info("消息成功发送并得到响应:{}", response);
if (response.getInteger("code") != 200) { if (response.getInteger("code") != 200) {
// TODO: 如果需要处理错误,您可以在这里添加逻辑 // TODO: 如果需要处理错误,您可以在这里添加逻辑
wxtsUtil.sendNotify("消息发送失败: " + responseStr);
throw new RuntimeException("消息发送失败: " + responseStr); throw new RuntimeException("消息发送失败: " + responseStr);
} }
//logger.info("消息成功发送并得到响应:{}", response); //logger.info("消息成功发送并得到响应:{}", response);
} else { } else {
wxtsUtil.sendNotify("消息发送失败,没有收到响应");
throw new RuntimeException("消息发送失败,没有收到响应"); throw new RuntimeException("消息发送失败,没有收到响应");
} }
} catch (Exception e) { } catch (Exception e) {

View File

@@ -42,7 +42,7 @@ import static cn.van.business.util.WXUtil.super_admins;
* @description * @description
*/ */
@Component @Component
public class JDUtils { public class JDUtil {
/** /**
* 密钥配置 * 密钥配置
*/ */
@@ -67,7 +67,7 @@ public class JDUtils {
//accessToken //accessToken
private static final String ACCESS_TOKEN = ""; private static final String ACCESS_TOKEN = "";
private static final Logger logger = LoggerFactory.getLogger(JDUtils.class); private static final Logger logger = LoggerFactory.getLogger(JDUtil.class);
private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
private final StringRedisTemplate redisTemplate; private final StringRedisTemplate redisTemplate;
@@ -77,7 +77,7 @@ public class JDUtils {
// 通过构造函数注入所有依赖项Spring将自动注入这些依赖 // 通过构造函数注入所有依赖项Spring将自动注入这些依赖
@Autowired // @Autowired 在构造函数上可以省略如果类只有一个构造函数 @Autowired // @Autowired 在构造函数上可以省略如果类只有一个构造函数
public JDUtils(StringRedisTemplate redisTemplate, OrderRowRepository orderRowRepository, WXUtil wxUtil, OrderUtil orderUtil) { public JDUtil(StringRedisTemplate redisTemplate, OrderRowRepository orderRowRepository, WXUtil wxUtil, OrderUtil orderUtil) {
this.redisTemplate = redisTemplate; this.redisTemplate = redisTemplate;
this.orderRowRepository = orderRowRepository; this.orderRowRepository = orderRowRepository;
this.wxUtil = wxUtil; this.wxUtil = wxUtil;
@@ -196,7 +196,7 @@ public class JDUtils {
if (Util.isAnyEmpty(appKey,secretKey)){ if (Util.isAnyEmpty(appKey,secretKey)){
continue; continue;
} }
logger.info("实时订单 {} ",appKey.substring(appKey.length()-4)); //logger.info("实时订单 {} ",appKey.substring(appKey.length()-4));
UnionOpenOrderRowQueryResponse response = fetchOrdersForDateTime(lastMinute, true, 1, true,appKey,secretKey); // 真实代表实时订单 UnionOpenOrderRowQueryResponse response = fetchOrdersForDateTime(lastMinute, true, 1, true,appKey,secretKey); // 真实代表实时订单
if (response != null) { if (response != null) {
@@ -565,7 +565,7 @@ public class JDUtils {
} }
// 打印方法调用和开始结束时间 // 打印方法调用和开始结束时间
logger.info(" {} --- 拉取订单, 分钟还是秒 {} , 开始时间:{} --- 结束时间:{}", appKey.substring(appKey.length()-4) ,hourMinuteTag, startTime.format(DATE_TIME_FORMATTER), endTime.format(DATE_TIME_FORMATTER)); //logger.info(" {} --- 拉取订单, 分钟还是秒 {} , 开始时间:{} --- 结束时间:{}", appKey.substring(appKey.length()-4) ,hourMinuteTag, startTime.format(DATE_TIME_FORMATTER), endTime.format(DATE_TIME_FORMATTER));
return unionOpenOrderRowQueryResponse; return unionOpenOrderRowQueryResponse;
} catch (Exception e) { } catch (Exception e) {

View File

@@ -31,6 +31,9 @@ public class WXUtil {
public static Map<String, SuperAdmin> super_admins = new HashMap<>(); public static Map<String, SuperAdmin> super_admins = new HashMap<>();
public static Map<String, String> jdidToWxidMap = new HashMap<>(); public static Map<String, String> jdidToWxidMap = new HashMap<>();
private final WxtsUtil wxTsUtil;
private final MessageProducerService messageProducerService;
@Data @Data
@AllArgsConstructor @AllArgsConstructor
@NoArgsConstructor @NoArgsConstructor
@@ -39,21 +42,21 @@ public class WXUtil {
private String name; private String name;
// 联盟ID // 联盟ID
private String unionId; private String unionId;
private String appKey ; private String appKey;
private String secretKey ; private String secretKey;
} }
// 初始化超级管理员 // 初始化超级管理员
public void initSuperAdmins() { public void initSuperAdmins() {
SuperAdmin admin1 = new SuperAdmin("wxid_ytpc72mdoskt22", "", "2014264913", SuperAdmin admin1 = new SuperAdmin("wxid_ytpc72mdoskt22", "", "2014264913", "98e21c89ae5610240ec3f5f575f86a59", "3dcb6b23a1104639ac433fd07adb6dfb");
"98e21c89ae5610240ec3f5f575f86a59","3dcb6b23a1104639ac433fd07adb6dfb");
super_admins.put(admin1.getWxid(), admin1); super_admins.put(admin1.getWxid(), admin1);
jdidToWxidMap.put(admin1.getUnionId(), admin1.getWxid()); jdidToWxidMap.put(admin1.getUnionId(), admin1.getWxid());
SuperAdmin admin2 = new SuperAdmin("wxid_yneqf1implxu12", "", "2025353364" SuperAdmin admin2 = new SuperAdmin("wxid_yneqf1implxu12", "", "2025353364", "e3c161242c8a1416fada5b5564d7ee70", "41ae9aabf03b41e6ba309682e36b323e");
,"e3c161242c8a1416fada5b5564d7ee70","41ae9aabf03b41e6ba309682e36b323e");
super_admins.put(admin2.getWxid(), admin2); super_admins.put(admin2.getWxid(), admin2);
jdidToWxidMap.put(admin2.getUnionId(), admin2.getWxid()); jdidToWxidMap.put(admin2.getUnionId(), admin2.getWxid());
wxTsUtil.sendNotify("initSuperAdmins 初始化完成");
// add more admins as needed... // add more admins as needed...
} }
@@ -94,11 +97,12 @@ public class WXUtil {
*/ */
public static String WX_BASE_URL; public static String WX_BASE_URL;
private Environment env; private Environment env;
@Autowired
private MessageProducerService messageProducerService;
@Autowired @Autowired
public WXUtil(Environment env) { public WXUtil(Environment env, WxtsUtil wxTsUtil, MessageProducerService messageProducerService) {
this.messageProducerService = messageProducerService;
this.wxTsUtil = wxTsUtil;
this.env = env; this.env = env;
WX_BASE_URL = env.getProperty("config.WX_BASE_URL"); WX_BASE_URL = env.getProperty("config.WX_BASE_URL");
System.out.println("WX_BASE_URL:" + WX_BASE_URL); System.out.println("WX_BASE_URL:" + WX_BASE_URL);

View File

@@ -6,6 +6,7 @@ import cn.van.business.model.wx.WxMessage;
import cn.van.business.repository.SettingRepository; import cn.van.business.repository.SettingRepository;
import cn.van.business.repository.WxMessageDataForChatRepository; import cn.van.business.repository.WxMessageDataForChatRepository;
import cn.van.business.repository.WxUserRepository; import cn.van.business.repository.WxUserRepository;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONArray;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -61,14 +62,14 @@ public class WxMessageConsumer {
private final WxMessageDataForChatRepository wxMessageDataForChatRepository; private final WxMessageDataForChatRepository wxMessageDataForChatRepository;
private final WxUserRepository wxUserRepository; private final WxUserRepository wxUserRepository;
private final SettingRepository settingRepository; private final SettingRepository settingRepository;
private final JDUtils jdUtils; private final JDUtil jdUtils;
@Autowired @Autowired
public WxMessageConsumer(WXUtil wxUtil, QLUtil qlUtil, public WxMessageConsumer(WXUtil wxUtil, QLUtil qlUtil,
@Lazy WxMessageDataForChatRepository wxMessageDataForChatService, @Lazy WxMessageDataForChatRepository wxMessageDataForChatService,
@Lazy WxUserRepository wxUserRepository, @Lazy WxUserRepository wxUserRepository,
@Lazy SettingRepository settingRepository, @Lazy SettingRepository settingRepository,
@Lazy JDUtils jdUtils) { @Lazy JDUtil jdUtils) {
this.wxUtil = wxUtil; this.wxUtil = wxUtil;
this.qlUtil = qlUtil; this.qlUtil = qlUtil;
this.wxMessageDataForChatRepository = wxMessageDataForChatService; this.wxMessageDataForChatRepository = wxMessageDataForChatService;
@@ -245,7 +246,7 @@ public class WxMessageConsumer {
private void handlePrivateMessage(WxMessage wxMessage) throws Exception { private void handlePrivateMessage(WxMessage wxMessage) throws Exception {
Integer msgType = 1; Integer msgType = 1;
// 做业务处理 // 做业务处理
//logger.info("处理消息: {}", JSON.toJSONString(wxMessage)); logger.info("处理消息: {}", JSON.toJSONString(wxMessage));
/** /**
* { * {

View File

@@ -0,0 +1,44 @@
package cn.van.business.util;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import java.util.HashMap;
/**
* @author Leo
* @version 1.0
* @create 2025/1/22 10:20
* @description 企业微信推送工具类
*/
@Component
public class WxtsUtil {
private static final Logger logger = LoggerFactory.getLogger(WxtsUtil.class);
public static final String TOKEN = "super_token_b62190c26";
private static final String SERVER_URL = "https://wxts.van333.cn";
public void sendNotify(String content) {
try {
String url = SERVER_URL + "/wx/send/jd";
HashMap<String, Object> paramMap = new HashMap<>();
paramMap.put("title", "JD机器人微信推送");
content = content.replaceAll("\\n", "<br>");
String common = "192.168.8.88 (微信机器人), 信息 : ";
content = common + content + "<br><br>";
paramMap.put("text", content);
HttpResponse execute = HttpRequest.post(url).header("vanToken", TOKEN).header("source", "XZJ_UBUNTU").body(JSON.toJSONString(paramMap)).execute();
logger.info("企业微信推送结果:{}", execute);
} catch (Exception e) {
logger.error("企业微信推送失败:{}", e.getMessage());
}
}
}