Compare commits

...

3 Commits

Author SHA1 Message Date
Leo
2316951e7f 重构第一版,没有明显bug 2025-03-11 14:11:51 +08:00
Leo
9aa02430fd 重构第一版,没有明显bug 2025-03-06 10:56:35 +08:00
Leo
645b025172 稳定版。没重构之前。稳定的限流。 2025-03-04 14:29:48 +08:00
27 changed files with 772 additions and 649 deletions

52
pom.xml
View File

@@ -15,9 +15,9 @@
<properties> <properties>
<java.version>17</java.version> <java.version>17</java.version>
<spring-boot.version>3.1.5</spring-boot.version> <spring-boot.version>3.1.5</spring-boot.version> <!-- 推荐使用最新3.1.x -->
<maven.compiler.source>17</maven.compiler.source> <rocketmq.version>2.3.2</rocketmq.version>
<maven.compiler.target>17</maven.compiler.target> <maven.compiler.release>17</maven.compiler.release>
</properties> </properties>
@@ -54,16 +54,14 @@
<artifactId>mysql-connector-j</artifactId> <artifactId>mysql-connector-j</artifactId>
<version>8.2.0</version> <version>8.2.0</version>
</dependency> </dependency>
<!-- 修改RocketMQ依赖 -->
<dependency> <dependency>
<groupId>org.apache.rocketmq</groupId> <groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId> <artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.1</version> <version>${rocketmq.version}</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>33.3.1-jre</version>
</dependency> </dependency>
<!-- Spring Boot Starter Test测试依赖 --> <!-- Spring Boot Starter Test测试依赖 -->
<dependency> <dependency>
<groupId>org.springframework.boot</groupId> <groupId>org.springframework.boot</groupId>
@@ -108,29 +106,38 @@
<artifactId>jdk</artifactId> <artifactId>jdk</artifactId>
<version>2.0</version> <version>2.0</version>
</dependency> </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<!-- pom.xml -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.junit.platform</groupId>
<artifactId>junit-platform-commons</artifactId>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<version>4.5.13</version>
</dependency>
</dependencies> </dependencies>
<!-- Maven 插件 --> <!-- Maven 插件 -->
<build> <build>
<plugins> <plugins>
<!-- 核心配置:必须启用 spring-boot-maven-plugin -->
<plugin> <plugin>
<groupId>org.springframework.boot</groupId> <groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId> <artifactId>spring-boot-maven-plugin</artifactId>
<version>3.1.5</version> <!-- 与你的Spring Boot版本一致 -->
<configuration> <configuration>
<!-- 显式指定主类(可选,建议添加) -->
<mainClass>cn.van.Application</mainClass> <mainClass>cn.van.Application</mainClass>
</configuration> </configuration>
<executions>
<execution>
<goals>
<goal>repackage</goal> <!-- 关键生成可执行JAR -->
</goals>
</execution>
</executions>
</plugin> </plugin>
</plugins> </plugins>
</build> </build>
@@ -140,5 +147,12 @@
<name>Local Repository</name> <name>Local Repository</name>
<url>http://192.168.8.88:8081/repository/maven-local88/</url> <url>http://192.168.8.88:8081/repository/maven-local88/</url>
</repository> </repository>
<!-- 在pom.xml的repositories节点中添加RocketMQ官方仓库 -->
<repository>
<id>rocketmq-repo</id>
<name>RocketMQ Repository</name>
<url>https://repo1.maven.org/maven2/org/apache/rocketmq/</url>
</repository>
</repositories> </repositories>
</project> </project>

View File

@@ -1,8 +1,10 @@
package cn.van; package cn.van;
import jakarta.annotation.PostConstruct; import jakarta.annotation.PostConstruct;
import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
import org.springframework.boot.SpringApplication; import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Import;
import org.springframework.core.env.Environment; import org.springframework.core.env.Environment;
import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.EnableScheduling;
@@ -16,6 +18,7 @@ import java.util.Arrays;
*/ */
@SpringBootApplication @SpringBootApplication
@EnableScheduling @EnableScheduling
@Import(RocketMQAutoConfiguration.class)
public class Application { public class Application {
private final Environment env; private final Environment env;

View File

@@ -1,15 +0,0 @@
package cn.van.business.config;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RocketMQConfig {
@Bean
public RocketMQTemplate rocketMQTemplate() {
System.out.println("RocketMQTemplate init");
return new RocketMQTemplate();
}
}

View File

@@ -1,12 +1,12 @@
package cn.van.business.config; package cn.van.business.config;
import jakarta.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.SchedulingConfigurer; import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar; import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import jakarta.annotation.PreDestroy; // 使用 jakarta.annotation 包
import java.util.Collection; import java.util.Collection;
import java.util.List; import java.util.List;
import java.util.concurrent.*; import java.util.concurrent.*;

View File

@@ -44,7 +44,7 @@ public enum WXReqType {
*/ */
GET_WX_LIST("getWeChatList", "获取微信列表"), GET_WX_LIST("getWeChatList", "获取微信列表"),
GET_WX_STATUS("checkWeChat", "微信状态检测"), GET_WX_STATUS("checkWeChat", "微信状态检测"),
SEND_TEXT_MESSAGE("sendText", "发送文本消息"), SEND_TEXT_MESSAGE("sendText2", "发送文本消息"),
UPDATE_DOWNLOAD_IMAGE("Q0002", "修改下载图片"), UPDATE_DOWNLOAD_IMAGE("Q0002", "修改下载图片"),
GET_USER_INFO("Q0003", "获取个人信息"), GET_USER_INFO("Q0003", "获取个人信息"),
QUERY_OBJECT_INFO("Q0004", "查询对象信"), QUERY_OBJECT_INFO("Q0004", "查询对象信"),

View File

@@ -8,7 +8,6 @@ package cn.van.business.model.jd;
*/ */
import jakarta.persistence.*; import jakarta.persistence.*;
import jakarta.persistence.TemporalType;
import java.util.Date; import java.util.Date;

View File

@@ -1,6 +1,7 @@
package cn.van.business.model.jd; package cn.van.business.model.jd;
import jakarta.persistence.*; import jakarta.persistence.*;
import java.util.Date; import java.util.Date;
/** /**

View File

@@ -2,8 +2,6 @@ package cn.van.business.model.jd;
import jakarta.persistence.*; import jakarta.persistence.*;
import java.util.Set;
/** /**
* sku对应的商品信息 * sku对应的商品信息
*/ */

View File

@@ -2,9 +2,6 @@ package cn.van.business.model.jd;
import jakarta.persistence.*; import jakarta.persistence.*;
import java.util.List;
import java.util.Set;
/** /**
* 实体类,用于存储商品类型信息。 * 实体类,用于存储商品类型信息。
*/ */

View File

@@ -2,10 +2,11 @@ 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.van.business.util.WxtsUtil; 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.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
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;
@@ -13,10 +14,8 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets; import java.util.concurrent.TimeUnit;
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;
/** /**
@@ -26,14 +25,17 @@ import static cn.van.business.util.WXUtil.WX_BASE_URL;
* @description * @description
*/ */
@Service @Service
@RocketMQMessageListener(topic = "wx-message", consumerGroup = "${rocketmq.consumer.group}", nameServer = "${rocketmq.name-server}") @RocketMQMessageListener(topic = "wx-message", consumerGroup = "${rocketmq.consumer.group}", nameServer = "${rocketmq.name-server}"
)
public class MessageConsumerService implements RocketMQListener<JSONObject> { 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 static final RateLimiter rateLimiter = RateLimiter.create(2, // 1 QPS
5, // 预热期 5 秒
TimeUnit.SECONDS);
private final WxtsUtil wxtsUtil; private final WxtsUtil wxtsUtil;
// create a rate limiter of 1 qps
RateLimiter rateLimiter = RateLimiter.create(0.5);
@Autowired @Autowired
public MessageConsumerService(WxtsUtil wxtsUtil) { public MessageConsumerService(WxtsUtil wxtsUtil) {
@@ -41,23 +43,23 @@ public class MessageConsumerService implements RocketMQListener<JSONObject> {
} }
@Override @Override
public void onMessage(JSONObject jsonObject) { public void onMessage(JSONObject message) {
// 处理消息
try { try {
rateLimiter.acquire(); // 请求许可。如果超过速率,则此方法会阻塞 //logger.info("[RateLimiter] 开始处理消息,当前时间:{}", System.currentTimeMillis());
String body = jsonObject.getString("body"); rateLimiter.acquire();
byte[] decodedBody = Base64.getDecoder().decode(body); //logger.info("[RateLimiter] 获得令牌,当前时间:{}", System.currentTimeMillis());
String decodedBodyStr = new String(decodedBody, StandardCharsets.UTF_8);
JSONObject decodedBodyJson = JSONObject.parseObject(decodedBodyStr);
String jsonStr = JSONUtil.toJsonStr(decodedBodyJson); //logger.debug("构造完成的消息结构:{}", requestBody.toJSONString());
String responseStr = HttpRequest.post(WX_BASE_URL)
.body(jsonStr) // 4. 发送请求(保持原有)
.execute() String responseStr;
.body(); responseStr = HttpRequest.post(WX_BASE_URL).body(message.toJSONString()).execute().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); //logger.info("消息成功发送并得到响应:{}", response);
if (response.getInteger("code") != 200) { if (response.getInteger("code") != 200) {
// TODO: 如果需要处理错误,您可以在这里添加逻辑 // TODO: 如果需要处理错误,您可以在这里添加逻辑
wxtsUtil.sendNotify("消息发送失败: " + responseStr); wxtsUtil.sendNotify("消息发送失败: " + responseStr);
@@ -69,9 +71,10 @@ public class MessageConsumerService implements RocketMQListener<JSONObject> {
throw new RuntimeException("消息发送失败,没有收到响应"); throw new RuntimeException("消息发送失败,没有收到响应");
} }
} catch (Exception e) { } catch (Exception e) {
//logger.error("处理消息时发生错误", e); logger.error("消息处理失败,原始消息:{}", message, e);
throw e; // 重抛异常使得 RocketMQ 可以捕获到这个异常 wxtsUtil.sendNotify("系统异常:" + e.getMessage());
}
} }
}
} }

View File

@@ -1,36 +1,56 @@
package cn.van.business.mq; package cn.van.business.mq;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import jakarta.annotation.PostConstruct;
import lombok.SneakyThrows; import lombok.SneakyThrows;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
/**
* @author Leo
* @version 1.0
* @create 2024/12/1 上午2:06
* @description
*/
@Service @Service
public class MessageProducerService { public class MessageProducerService {
private static final Logger logger = LoggerFactory.getLogger(MessageProducerService.class); private static final Logger logger = LoggerFactory.getLogger(MessageProducerService.class);
private static final String topic = "wx-message"; private static final String topic = "wx-message";
private final RocketMQTemplate rocketMQTemplate; private final RocketMQTemplate rocketMQTemplate;
public MessageProducerService(RocketMQTemplate rocketMQTemplate) { public MessageProducerService(RocketMQTemplate rocketMQTemplate
) {
this.rocketMQTemplate = rocketMQTemplate; this.rocketMQTemplate = rocketMQTemplate;
} }
@PostConstruct
public void init() {
if (rocketMQTemplate == null) {
throw new IllegalStateException("RocketMQTemplate not initialized!");
}
}
@SneakyThrows @SneakyThrows
public void sendMessage(JSONObject jsonObject) { public void sendMessage(JSONObject data) {
Message message = new Message(topic, jsonObject.toJSONString().getBytes()); // 消息结构校验
message.setTags("wx"); if (!data.containsKey("type") || !data.containsKey("data")) {
rocketMQTemplate.convertAndSend(topic, message); logger.error("非法消息格式:{}", data);
throw new IllegalArgumentException("消息必须包含type和data字段");
}
// 新增校验
if (!data.getJSONObject("data").containsKey("wxid")) {
throw new IllegalArgumentException("消息必须包含wxid字段");
}
// 构建Spring Message
Message<String> message = MessageBuilder
.withPayload(data.toJSONString())
.setHeader(RocketMQHeaders.TAGS, "wx")
.build();
// 发送消息
rocketMQTemplate.send(topic, message);
logger.debug("消息已发送:{}", data);
} }
} }

View File

@@ -8,8 +8,11 @@ package cn.van.business.repository;
*/ */
import cn.van.business.model.jd.OrderRow; import cn.van.business.model.jd.OrderRow;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Query; import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;
import org.springframework.format.annotation.DateTimeFormat; import org.springframework.format.annotation.DateTimeFormat;
import org.springframework.stereotype.Repository; import org.springframework.stereotype.Repository;
@@ -34,8 +37,13 @@ public interface OrderRowRepository extends JpaRepository<OrderRow, String> {
List<OrderRow> findByValidCodeNotInOrderByOrderTimeDescAndUnionId(int[] validCodes, Long unionId); List<OrderRow> findByValidCodeNotInOrderByOrderTimeDescAndUnionId(int[] validCodes, Long unionId);
@Query("select o from OrderRow o where o.validCode not in ?1 and o.orderTime >= ?2 order by o.orderTime DESC") @Query("select o from OrderRow o where o.validCode not in ?1 and o.orderTime >= ?2 order by o.orderTime DESC")
List<OrderRow> findByValidCodeNotInAndOrderTimeGreaterThanOrderByOrderTimeDesc( List<OrderRow> findByValidCodeNotInAndOrderTimeGreaterThanOrderByOrderTimeDesc(int[] validCodes, @DateTimeFormat(pattern = "yyyy-MM-dd") Date threeMonthsAgo);
int[] validCodes,
@DateTimeFormat(pattern = "yyyy-MM-dd") Date threeMonthsAgo @Query("select o from OrderRow o where o.validCode not in ?1 and o.skuId = ?2 and o.unionId = ?3 order by o.orderTime DESC")
); List<OrderRow> findBySkuIdAndUnionId(int[] validCodes,long skuId, long unionId);
//// 在OrderRowRepository中添加模糊查询方法
//// 模糊查询收件人姓名或地址(包含分页)
//@Query("SELECT o FROM OrderRow o WHERE " + "o.recipientName LIKE %:keyword% OR " + "o.address LIKE %:keyword% " + "ORDER BY o.orderTime DESC")
//Page<OrderRow> searchByRecipientOrAddress(@Param("keyword") String keyword, Pageable pageable);
} }

View File

@@ -6,7 +6,9 @@ import java.sql.Timestamp;
import java.text.DateFormat; import java.text.DateFormat;
import java.text.ParseException; import java.text.ParseException;
import java.text.SimpleDateFormat; import java.text.SimpleDateFormat;
import java.util.*; import java.util.Calendar;
import java.util.Date;
import java.util.Locale;
/** /**
* 日期工具类 * 日期工具类

File diff suppressed because it is too large Load Diff

View File

@@ -56,10 +56,8 @@ public class WXUtil {
SuperAdmin admin2 = new SuperAdmin("wxid_yneqf1implxu12", "", "2025353364", "e3c161242c8a1416fada5b5564d7ee70", "41ae9aabf03b41e6ba309682e36b323e"); SuperAdmin admin2 = new SuperAdmin("wxid_yneqf1implxu12", "", "2025353364", "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 初始化完成"); //wxTsUtil.sendNotify("initSuperAdmins 初始化完成");
// add more admins as needed...
} }
public static String getWxidFromJdid(String jdid) { public static String getWxidFromJdid(String jdid) {
@@ -145,7 +143,7 @@ public class WXUtil {
//if (wxid.equals(super_admin_wxid) || fromwxid.equals(super_admin_wxid)) { //if (wxid.equals(super_admin_wxid) || fromwxid.equals(super_admin_wxid)) {
// content = "超管: 凡神 \r\n" + content; // content = "超管: 凡神 \r\n" + content;
//} //}
List<String> strings = splitStringByLength(content, 2048); List<String> strings = splitStringByLength(content, 4096);
int count = 1; int count = 1;
for (String string : strings) { for (String string : strings) {

View File

@@ -2,7 +2,6 @@ package cn.van.business.util;
import cn.hutool.http.HttpRequest; import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse; import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -33,7 +32,7 @@ public class WxtsUtil {
content = common + content + "<br><br>"; content = common + content + "<br><br>";
paramMap.put("text", content); paramMap.put("text", content);
HttpResponse execute = HttpRequest.post(url).header("vanToken", TOKEN).header("source", "XZJ_UBUNTU").body(JSON.toJSONString(paramMap)).execute(); HttpResponse execute = HttpRequest.post(url).header("vanToken", TOKEN).header("source", "XZJ_UBUNTU").body(JSON.toJSONString(paramMap)).execute();
logger.info("企业微信推送结果:{}", execute); //logger.info("企业微信推送结果:{}", execute);
} catch (Exception e) { } catch (Exception e) {
logger.error("企业微信推送失败:{}", e.getMessage()); logger.error("企业微信推送失败:{}", e.getMessage());
} }
@@ -41,4 +40,13 @@ public class WxtsUtil {
} }
// 添加分级告警方法
public void sendCriticalAlert(String title, String content) {
String formattedMsg = String.format("[CRITICAL] %s\n%s", title, content);
// 这里调用实际的通知渠道,例如:
// - 发送邮件
// - 调用企业微信机器人
// - 触发短信通知
sendNotify(formattedMsg); // 复用原有通知方法
}
} }

View File

@@ -0,0 +1,26 @@
package cn.van.business.util.jdReq;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
@Component
public class Days0007Strategy implements OrderFetchStrategy {
@Override
public TimeRange calculateRange(LocalDateTime baseTime) {
LocalDateTime end = baseTime.truncatedTo(ChronoUnit.HOURS);
LocalDateTime start = end.minusDays(7).truncatedTo(ChronoUnit.HOURS);
if (start.isAfter(end)) { // 防御性校验
throw new IllegalArgumentException("时间范围错误");
}
return new TimeRange(start, end);
}
@Override
public String strategyName() {
return "00-07天历史订单抓取策略";
}
}

View File

@@ -0,0 +1,25 @@
package cn.van.business.util.jdReq;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
@Component
public class Days0714Strategy implements OrderFetchStrategy {
@Override
public TimeRange calculateRange(LocalDateTime baseTime) {
LocalDateTime end = baseTime.truncatedTo(ChronoUnit.HOURS).minusDays(7);
LocalDateTime start = end.minusDays(14).truncatedTo(ChronoUnit.HOURS);
if (start.isAfter(end)) { // 防御性校验
throw new IllegalArgumentException("时间范围错误");
}
return new TimeRange(start, end);
}
@Override
public String strategyName() {
return "07-14天历史订单抓取策略";
}
}

View File

@@ -0,0 +1,24 @@
package cn.van.business.util.jdReq;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
// 在jdReq包中补充策略类
public class Days1430Strategy implements OrderFetchStrategy {
@Override
public TimeRange calculateRange(LocalDateTime baseTime) {
LocalDateTime end = baseTime.minusDays(30).truncatedTo(ChronoUnit.HOURS);
LocalDateTime start = baseTime.minusDays(14).truncatedTo(ChronoUnit.HOURS);
if (start.isAfter(end)) { // 防御性校验
throw new IllegalArgumentException("时间范围错误");
}
return new TimeRange(start, end);
}
@Override
public String strategyName() {
return "14-30天历史订单抓取策略";
}
}
// 其他策略类类似实现

View File

@@ -0,0 +1,24 @@
package cn.van.business.util.jdReq;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
@Component
public class Days3090Strategy implements OrderFetchStrategy {
@Override
public TimeRange calculateRange(LocalDateTime baseTime) {
LocalDateTime end = baseTime.minusMonths(1);
LocalDateTime start = end.minusMonths(2);
if (start.isAfter(end)) { // 防御性校验
throw new IllegalArgumentException("时间范围错误");
}
return new TimeRange(start, end);
}
@Override
public String strategyName() {
return "30-90天历史订单抓取策略";
}
}

View File

@@ -0,0 +1,40 @@
//package cn.van.business.util.jdReq;
//
//import cn.van.business.repository.OrderRowRepository;
//import com.jd.open.api.sdk.response.kplunion.UnionOpenOrderRowQueryResponse;
//import org.springframework.beans.factory.annotation.Autowired;
//
//import java.time.LocalDateTime;
//
//public abstract class HistoricalOrderFetcher {
// @Autowired
// protected OrderRowRepository orderRowRepository;
//
// protected int fetchOrders(OrderFetchStrategy strategy) {
// int count = 0;
// LocalDateTime start = strategy.getStartTime();
// LocalDateTime end = strategy.getEndTime();
//
// while (!start.isEqual(end)) {
// Integer pageIndex = 1;
// boolean hasMore;
//
// do {
// UnionOpenOrderRowQueryResponse response = fetchPage(strategy, start, pageIndex);
// hasMore = processResponse(response, strategy);
// pageIndex++;
// } while (hasMore);
//
// start = start.plusHours(1);
// }
// return count;
// }
//
// protected abstract UnionOpenOrderRowQueryResponse fetchPage(OrderFetchStrategy strategy,
// LocalDateTime startTime,
// Integer pageIndex);
//
// private boolean processResponse(UnionOpenOrderRowQueryResponse response, OrderFetchStrategy strategy) {
// // 统一响应处理逻辑...
// }
//}

View File

@@ -0,0 +1,18 @@
package cn.van.business.util.jdReq;
import java.time.LocalDateTime;
public interface OrderFetchStrategy {
/**
* 计算要抓取的时间范围
* @param baseTime 基准时间(通常用当前时间)
* @return 包含开始时间和结束时间的值对象
*/
TimeRange calculateRange(LocalDateTime baseTime);
/**
* 策略标识
*/
String strategyName();
}

View File

@@ -0,0 +1,11 @@
package cn.van.business.util.jdReq;
public class StrategyFactory {
public static OrderFetchStrategy getStrategy(String type) {
switch (type) {
case "30-90": return new Days3090Strategy();
//case "14-30": return new Days1430Strategy();
default: throw new IllegalArgumentException();
}
}
}

View File

@@ -0,0 +1,14 @@
package cn.van.business.util.jdReq;
import lombok.AllArgsConstructor;
import lombok.Getter;
import java.time.LocalDateTime;
// 时间范围值对象
@Getter
@AllArgsConstructor
public class TimeRange {
private LocalDateTime start;
private LocalDateTime end;
}

View File

@@ -60,3 +60,6 @@ rocketmq:
consume-thread-min: 20 # 消费线程池最小线程数 consume-thread-min: 20 # 消费线程池最小线程数
consume-thread-max: 64 # 消费线程池最大线程数 consume-thread-max: 64 # 消费线程池最大线程数
consume-message-batch-max-size: 64 # 批量消费最大消息数 consume-message-batch-max-size: 64 # 批量消费最大消息数
isRunning:
wx: false
jd: false

View File

@@ -60,3 +60,6 @@ rocketmq:
consume-thread-min: 20 # 消费线程池最小线程数 consume-thread-min: 20 # 消费线程池最小线程数
consume-thread-max: 64 # 消费线程池最大线程数 consume-thread-max: 64 # 消费线程池最大线程数
consume-message-batch-max-size: 64 # 批量消费最大消息数 consume-message-batch-max-size: 64 # 批量消费最大消息数
isRunning:
wx: true
jd: true

View File

@@ -61,3 +61,21 @@ rocketmq:
consume-thread-min: 20 # 消费线程池最小线程数 consume-thread-min: 20 # 消费线程池最小线程数
consume-thread-max: 64 # 消费线程池最大线程数 consume-thread-max: 64 # 消费线程池最大线程数
consume-message-batch-max-size: 64 # 批量消费最大消息数 consume-message-batch-max-size: 64 # 批量消费最大消息数
management:
endpoints:
web:
exposure:
include: health,metrics,resilience4j
prometheus:
metrics:
export:
enabled: true
resilience4j.ratelimiter:
instances:
wxMsgLimiter:
limitForPeriod: 10 # 根据业务吞吐量调整
limitRefreshPeriod: 1s # 固定1秒周期
timeoutDuration: 0 # 立即失败模式
registerHealthIndicator: true