From 228c4cf732b8883e741dbd2242207419aacc58b6 Mon Sep 17 00:00:00 2001 From: cst61 Date: Tue, 2 Jun 2026 19:16:10 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AF=B7=E8=B4=AD=E6=88=90=E5=8A=9F=E7=9A=84?= =?UTF-8?q?=E7=89=88=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/rj/controller/LbGoodsController.java | 2 +- .../com/rj/mapper/LbBuyAccountMapper.java | 21 ++ .../rj/scheduler/LBAdminPullScheduler.java | 149 ++++++++++ .../java/com/rj/service/ILbGoodsService.java | 16 +- .../service/impl/LbBuyAccountServiceImpl.java | 271 +++++++++++++----- .../rj/service/impl/LbGoodsServiceImpl.java | 153 +++++++++- 6 files changed, 533 insertions(+), 79 deletions(-) diff --git a/src/main/java/com/rj/controller/LbGoodsController.java b/src/main/java/com/rj/controller/LbGoodsController.java index bd96a60..bf48ce1 100644 --- a/src/main/java/com/rj/controller/LbGoodsController.java +++ b/src/main/java/com/rj/controller/LbGoodsController.java @@ -139,7 +139,7 @@ public class LbGoodsController { @Operation( summary = "抢购货品", description = - "从 lb_goods 按 total_money 从大到小选取金额小于39000的货品，调用 hxrd POST /api/order/buy（body: id、seller_id）；" + "从 lb_goods 选取金额大于29000且24小时内更新的货品，调用 hxrd POST /api/order/buy（body: id、seller_id）；" + "成功笔数达到 maxBuyCount 后停止，且循环抢购总次数不超过 maxBuyCount 的5倍；" + "token、URL、appStr 等从 lb_third_integration_config 按 tenantId 读取；" + "token 可选，传入时覆盖库中 goodsApiToken") diff --git a/src/main/java/com/rj/mapper/LbBuyAccountMapper.java b/src/main/java/com/rj/mapper/LbBuyAccountMapper.java index ebb8bb3..126acd7 100644 --- a/src/main/java/com/rj/mapper/LbBuyAccountMapper.java +++ b/src/main/java/com/rj/mapper/LbBuyAccountMapper.java @@ -1,9 +1,30 @@ package com.rj.mapper; +import com.baomidou.mybatisplus.annotation.InterceptorIgnore; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.rj.entity.LbBuyAccount; import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Update; + +import java.time.LocalDateTime; @Mapper public interface LbBuyAccountMapper extends BaseMapper { + + /** + * 按主键更新抢单结果与时间；跳过多租户拦截，避免批量抢单 worker 线程无租户上下文导致更新 0 行。 + */ + @InterceptorIgnore(tenantLine = "true") + @Update(""" + UPDATE lb_buy_account + SET rush_buy_result = #{rushBuyResult}, + last_rush_buy_time = #{lastRushBuyTime}, + update_time = #{updateTime} + WHERE id = #{id} + """) + int updateRushBuyOutcome(@Param("id") String id, + @Param("rushBuyResult") String rushBuyResult, + @Param("lastRushBuyTime") LocalDateTime lastRushBuyTime, + @Param("updateTime") LocalDateTime updateTime); } diff --git a/src/main/java/com/rj/scheduler/LBAdminPullScheduler.java b/src/main/java/com/rj/scheduler/LBAdminPullScheduler.java index 1d04faa..c7e570c 100644 --- a/src/main/java/com/rj/scheduler/LBAdminPullScheduler.java +++ b/src/main/java/com/rj/scheduler/LBAdminPullScheduler.java @@ -2,14 +2,23 @@ package com.rj.scheduler; import com.rj.config.AppConfig; import com.rj.config.HxrAdminProperties; +import com.rj.entity.LbThirdIntegrationConfig; +import com.rj.entity.LbGoods; import com.rj.service.HxrAdminOrderSelectService; +import com.rj.service.ILbGoodsService; import com.rj.service.ILbOrderRowService; +import com.rj.service.ILbThirdIntegrationConfigService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; +import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.time.Clock; import java.time.DayOfWeek; @@ -29,6 +38,9 @@ public class LBAdminPullScheduler { private static final long FIFTY_MINUTES_MS = 30L * 60L * 1000L; private static final long UnPay_MINUTES_MS = 20L * 60L * 1000L; + private static final long GOODS_PULL_INTERVAL_MS = 1000L; + private static final LocalTime GOODS_PULL_WINDOW_START = LocalTime.of(9, 59); + private static final LocalTime GOODS_PULL_WINDOW_END = LocalTime.of(22, 3, 59); private static final DateTimeFormatter BUY_TIME_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @@ -36,8 +48,16 @@ public class LBAdminPullScheduler { private final HxrAdminProperties hxrAdminProperties; private final HxrAdminOrderSelectService hxrAdminOrderSelectService; private final ILbOrderRowService lbOrderRowService; + private final ILbThirdIntegrationConfigService lbThirdIntegrationConfigService; + private final ILbGoodsService lbGoodsService; private final Clock clock; + /** 防止货品拉取上一轮未完成时重复执行。 */ + private final AtomicBoolean goodsPullRunning = new AtomicBoolean(false); + + /** 最近一次调度拉取的货品：租户 ID -> 货品列表（线程安全）。 */ + private final ConcurrentHashMap> tenantGoodsCache = new ConcurrentHashMap<>(); + @Scheduled( fixedRate = FIFTY_MINUTES_MS, initialDelayString = "${hxr.admin.pull-initial-delay-ms:0}") @@ -147,5 +167,134 @@ public class LBAdminPullScheduler { } } + /** + * 拉取抢购的物品 + * 货品同步：每 1 秒触发一次，仅在工作日 9:59～20:03（应用时区）窗口内真正执行； + * 若上一轮尚未结束则跳过，避免重复并发执行。 + */ +// @Scheduled( +// fixedRate = GOODS_PULL_INTERVAL_MS, +// initialDelayString = "${hxr.admin.pull-initial-delay-ms:0}") + public void pullGoodsFromThirdParty() { + try { + + ZoneId zone = ZoneId.of(appConfig.getTimezone()); + if (!isWithinGoodsPullWindow(zone)) { + log.debug( + "当前不在货品拉取窗口（周一至周五 {}～{}），跳过", + GOODS_PULL_WINDOW_START, + GOODS_PULL_WINDOW_END); + return; + } + if (!goodsPullRunning.compareAndSet(false, true)) { + log.debug("货品拉取正在运行中，跳过重复执行"); + return; + } + try { + syncEnabledGoodsConfigs(zone); + } finally { + goodsPullRunning.set(false); + } + } catch (Exception e) { + goodsPullRunning.set(false); + log.error("货品拉取调度失败", e); + } + } + + private boolean isWithinGoodsPullWindow(ZoneId zone) { + ZonedDateTime nowZdt = ZonedDateTime.now(clock.withZone(zone)); + DayOfWeek dow = nowZdt.getDayOfWeek(); + if (dow == DayOfWeek.SATURDAY || dow == DayOfWeek.SUNDAY) { + return false; + } + LocalTime now = nowZdt.toLocalTime(); + return !now.isBefore(GOODS_PULL_WINDOW_START) && !now.isAfter(GOODS_PULL_WINDOW_END); + } + + private void syncEnabledGoodsConfigs(ZoneId zone) { + List configs = lbThirdIntegrationConfigService.lambdaQuery() + .eq(LbThirdIntegrationConfig::getEnabled, 1) + .list(); + if (configs == null || configs.isEmpty()) { + log.debug("无启用的第三方集成配置，跳过货品拉取"); + return; + } + if (!isWithinGoodsPullWindow(zone)) { + log.debug("货品拉取窗口已结束，跳过本次调度"); + return; + } + + log.info("货品拉取开始，启用租户数={}", configs.size()); + ConcurrentHashMap> runGoodsMap = new ConcurrentHashMap<>(); + ConcurrentHashMap> runResultMap = new ConcurrentHashMap<>(); + AtomicInteger skippedCount = new AtomicInteger(0); + + List> futures = configs.stream() + .map(config -> CompletableFuture.runAsync(() -> { + String tenantId = config.getTenantId(); + if (tenantId == null || tenantId.isBlank()) { + log.warn("集成配置 id={} 缺少 tenantId，跳过", config.getId()); + skippedCount.incrementAndGet(); + return; + } + String tid = tenantId.trim(); + Map fetchResult = lbGoodsService.fetchGoodsFromHxrInMemory(tid, null); + runResultMap.put(tid, fetchResult); + if (Boolean.TRUE.equals(fetchResult.get("success"))) { + @SuppressWarnings("unchecked") + List goods = (List) fetchResult.get("data"); + runGoodsMap.put(tid, goods != null ? goods : List.of()); + } else { + runGoodsMap.put(tid, List.of()); + } + })) + .toList(); + + CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join(); + + tenantGoodsCache.clear(); + tenantGoodsCache.putAll(runGoodsMap); + + int successCount = 0; + int failCount = 0; + int totalGoods = 0; + StringBuilder detail = new StringBuilder(); + for (Map.Entry> entry : runResultMap.entrySet()) { + String tid = entry.getKey(); + Map fetchResult = entry.getValue(); + boolean success = Boolean.TRUE.equals(fetchResult.get("success")); + int fetched = fetchResult.get("fetched") instanceof Number n ? n.intValue() : 0; + if (success) { + successCount++; + totalGoods += fetched; + } else { + failCount++; + } + detail.append(System.lineSeparator()) + .append(" tenantId=") + .append(tid) + .append(" success=") + .append(success) + .append(" fetched=") + .append(fetched) + .append(" message=") + .append(fetchResult.get("message")); + } + + log.info( + "货品拉取调度完成 配置租户数={} 跳过={} 成功={} 失败={} 内存租户数={} 总货品数={}{}", + configs.size(), + skippedCount.get(), + successCount, + failCount, + tenantGoodsCache.size(), + totalGoods, + detail); + } + + /** 获取最近一次调度写入内存的租户货品映射（只读视图）。 */ + public Map> getTenantGoodsCache() { + return Map.copyOf(tenantGoodsCache); + } } diff --git a/src/main/java/com/rj/service/ILbGoodsService.java b/src/main/java/com/rj/service/ILbGoodsService.java index 8b39710..13e9f93 100644 --- a/src/main/java/com/rj/service/ILbGoodsService.java +++ b/src/main/java/com/rj/service/ILbGoodsService.java @@ -37,10 +37,22 @@ public interface ILbGoodsService extends IService { Map syncFromHxrGoods(String tenantId, String token); /** - * 按 {@code total_money} 从大到小选取金额小于39000的 {@code lb_goods}，调用 hxrd {@code /api/order/buy} 抢购； + * 分页拉取第三方 {@code /api/order/goods}，仅组装为内存对象列表，不写入 {@code lb_goods}。 + * + * @param token 可选；非空时作为请求头 token，否则使用集成配置中的 goodsApiToken + */ + Map fetchGoodsFromHxrInMemory(String tenantId, String token); + + /** + * 选取金额大于29000且24小时内更新的 {@code lb_goods}，调用 hxrd {@code /api/order/buy} 抢购； * 成功笔数达到 {@code maxBuyCount} 后停止，且循环抢购总次数不超过 {@code maxBuyCount} 的5倍。 * * @param token 可选；非空时作为请求头 token，否则使用 {@code lb_third_integration_config} 中的 goodsApiToken + * @param rushBuyAccountLabel 可选；批量账号抢单时传入，用于日志标识当前抢单账号 */ - Map rushBuy(String tenantId, String token, Integer maxBuyCount); + default Map rushBuy(String tenantId, String token, Integer maxBuyCount) { + return rushBuy(tenantId, token, maxBuyCount, null); + } + + Map rushBuy(String tenantId, String token, Integer maxBuyCount, String rushBuyAccountLabel); } diff --git a/src/main/java/com/rj/service/impl/LbBuyAccountServiceImpl.java b/src/main/java/com/rj/service/impl/LbBuyAccountServiceImpl.java index 9f04186..7bed786 100644 --- a/src/main/java/com/rj/service/impl/LbBuyAccountServiceImpl.java +++ b/src/main/java/com/rj/service/impl/LbBuyAccountServiceImpl.java @@ -19,6 +19,11 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; @Slf4j @Service @@ -226,75 +231,96 @@ public class LbBuyAccountServiceImpl } } - List> accountResults = new ArrayList<>(); - int successAccounts = 0; - int failAccounts = 0; + int accountCount = normalizedIds.size(); + AtomicInteger threadSeq = new AtomicInteger(0); + ExecutorService batchExecutor = Executors.newFixedThreadPool(accountCount, r -> { + Thread t = new Thread(r, "lb-rush-buy-" + threadSeq.incrementAndGet()); + t.setDaemon(true); + return t; + }); + log.info("批量抢单启动，账号数={}，独立线程数={}", accountCount, accountCount); - for (String id : normalizedIds) { - Map item = new LinkedHashMap<>(); - item.put("accountId", id); + try { + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch readyLatch = new CountDownLatch(accountCount); + CountDownLatch doneLatch = new CountDownLatch(accountCount); + @SuppressWarnings("unchecked") + Map[] resultArray = new Map[accountCount]; - LbBuyAccount account = accountMap.get(id); - if (account == null) { - item.put("success", false); - item.put("message", "抢单账号不存在"); - accountResults.add(item); - failAccounts++; - continue; + for (int i = 0; i < accountCount; i++) { + final int index = i; + String id = normalizedIds.get(i); + LbBuyAccount account = accountMap.get(id); + batchExecutor.submit(() -> { + readyLatch.countDown(); + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + resultArray[index] = buildRushBuyTaskErrorItem(id, account, "抢单被中断"); + persistRushBuyOutcomeFromItem(account, resultArray[index]); + doneLatch.countDown(); + return; + } + try { + resultArray[index] = processSingleAccountRushBuy(id, account); + } catch (Exception e) { + log.error("抢单账号任务执行异常，accountId={}，loginAccount={}", + id, account != null ? account.getLoginAccount() : null, e); + resultArray[index] = buildRushBuyTaskErrorItem(id, account, e.getMessage()); + persistRushBuyOutcomeFromItem(account, resultArray[index]); + } finally { + doneLatch.countDown(); + } + }); } - item.put("loginAccount", account.getLoginAccount()); - item.put("tenantId", account.getTenantId()); - item.put("nickname", account.getNickname()); + if (!readyLatch.await(30, TimeUnit.SECONDS)) { + result.put("success", false); + result.put("message", "抢单线程就绪超时"); + return result; + } + log.info("全部 {} 个抢单线程已就绪，同时启动", accountCount); + startLatch.countDown(); - String resultMessage; - Boolean rushSuccess = false; - LocalDateTime rushBuyTime = LocalDateTime.now(); + if (!doneLatch.await(120, TimeUnit.SECONDS)) { + result.put("success", false); + result.put("message", "批量抢单执行超时"); + return result; + } - if (account.getEnabled() != null && account.getEnabled() == 0) { - resultMessage = formatPreRushBuyFailure("账号未启用"); - } else if (account.getTenantId() == null || account.getTenantId().trim().isEmpty()) { - resultMessage = formatPreRushBuyFailure("tenantId不能为空"); - } else if (account.getTokenFront() == null || account.getTokenFront().trim().isEmpty()) { - resultMessage = formatPreRushBuyFailure("token_front不能为空"); - } else { - Integer maxGrabCount = account.getMaxGrabCount(); - if (maxGrabCount == null || maxGrabCount <= 0) { - resultMessage = formatPreRushBuyFailure("maxGrabCount必须大于0"); + List> accountResults = new ArrayList<>(accountCount); + int successAccounts = 0; + int failAccounts = 0; + for (Map item : resultArray) { + accountResults.add(item); + if (Boolean.TRUE.equals(item.get("success"))) { + successAccounts++; } else { - Map rushBuyResult = lbGoodsService.rushBuy( - account.getTenantId().trim(), - account.getTokenFront().trim(), - maxGrabCount); - item.put("rushBuyResult", rushBuyResult); - rushSuccess = rushBuyResult.get("success") instanceof Boolean - ? (Boolean) rushBuyResult.get("success") - : null; - resultMessage = formatRushBuyResultMessage(rushBuyResult); + failAccounts++; } } - item.put("success", Boolean.TRUE.equals(rushSuccess)); - item.put("message", resultMessage); - accountResults.add(item); - persistRushBuyOutcome(account.getId(), resultMessage, rushBuyTime); - - if (Boolean.TRUE.equals(rushSuccess)) { - successAccounts++; - } else { - failAccounts++; + result.put("success", successAccounts > 0); + result.put("message", successAccounts > 0 + ? "批量抢购完成，成功账号 " + successAccounts + " 个，失败 " + failAccounts + " 个" + : "批量抢购未成功"); + result.put("totalAccounts", accountCount); + result.put("successAccounts", successAccounts); + result.put("failAccounts", failAccounts); + result.put("accountResults", accountResults); + return result; + } finally { + batchExecutor.shutdown(); + try { + if (!batchExecutor.awaitTermination(120, TimeUnit.SECONDS)) { + batchExecutor.shutdownNow(); + } + } catch (InterruptedException e) { + batchExecutor.shutdownNow(); + Thread.currentThread().interrupt(); } } - - result.put("success", successAccounts > 0); - result.put("message", successAccounts > 0 - ? "批量抢购完成，成功账号 " + successAccounts + " 个，失败 " + failAccounts + " 个" - : "批量抢购未成功"); - result.put("totalAccounts", normalizedIds.size()); - result.put("successAccounts", successAccounts); - result.put("failAccounts", failAccounts); - result.put("accountResults", accountResults); - return result; } catch (Exception e) { log.error("抢单账号批量抢购异常", e); result.put("success", false); @@ -303,6 +329,88 @@ public class LbBuyAccountServiceImpl } } + private static Map buildRushBuyTaskErrorItem(String accountId, LbBuyAccount account, String reason) { + String accountLabel = formatRushBuyAccountLabel(accountId, account); + Map item = new LinkedHashMap<>(); + item.put("accountId", accountId); + if (account != null) { + item.put("loginAccount", account.getLoginAccount()); + item.put("tenantId", account.getTenantId()); + item.put("nickname", account.getNickname()); + } + item.put("success", false); + item.put("message", attachAccountLabel(accountLabel, "抢单任务异常：" + reason)); + return item; + } + + /** + * 单账号抢单逻辑，供线程池并行调用。 + */ + private Map processSingleAccountRushBuy(String id, LbBuyAccount account) { + String accountLabel = formatRushBuyAccountLabel(id, account); + log.info("抢单任务开始，{}，thread={}", accountLabel, Thread.currentThread().getName()); + Map item = new LinkedHashMap<>(); + item.put("accountId", id); + + if (account == null) { + String message = attachAccountLabel(accountLabel, formatPreRushBuyFailure("抢单账号不存在")); + item.put("success", false); + item.put("message", message); + log.info("抢单任务结束，{}，success=false，message={}", accountLabel, message); + return item; + } + + item.put("loginAccount", account.getLoginAccount()); + item.put("tenantId", account.getTenantId()); + item.put("nickname", account.getNickname()); + + String resultMessage; + Boolean rushSuccess = false; + + if (account.getEnabled() != null && account.getEnabled() == 0) { + resultMessage = attachAccountLabel(accountLabel, formatPreRushBuyFailure("账号未启用")); + } else if (account.getTenantId() == null || account.getTenantId().trim().isEmpty()) { + resultMessage = attachAccountLabel(accountLabel, formatPreRushBuyFailure("tenantId不能为空")); + } else if (account.getTokenFront() == null || account.getTokenFront().trim().isEmpty()) { + resultMessage = attachAccountLabel(accountLabel, formatPreRushBuyFailure("token_front不能为空")); + } else { + Integer maxGrabCount = account.getMaxGrabCount(); + if (maxGrabCount == null || maxGrabCount <= 0) { + resultMessage = attachAccountLabel(accountLabel, formatPreRushBuyFailure("maxGrabCount必须大于0")); + } else { + Map rushBuyResult = lbGoodsService.rushBuy( + account.getTenantId().trim(), + account.getTokenFront().trim(), + maxGrabCount, + accountLabel); + item.put("rushBuyResult", rushBuyResult); + rushSuccess = rushBuyResult.get("success") instanceof Boolean + ? (Boolean) rushBuyResult.get("success") + : null; + resultMessage = attachAccountLabel( + accountLabel, formatRushBuyResultMessage(rushBuyResult)); + } + } + + item.put("success", Boolean.TRUE.equals(rushSuccess)); + item.put("message", resultMessage); + log.info("抢单任务结束，{}，success={}，message={}", accountLabel, item.get("success"), resultMessage); + persistRushBuyOutcome(account.getId(), accountLabel, resultMessage); + return item; + } + + private void persistRushBuyOutcomeFromItem(LbBuyAccount account, Map item) { + if (account == null || account.getId() == null || item == null) { + return; + } + String accountLabel = formatRushBuyAccountLabel(account.getId(), account); + Object message = item.get("message"); + persistRushBuyOutcome( + account.getId(), + accountLabel, + message != null ? message.toString() : "未知抢单结果"); + } + private Map validateRequiredForAdd(LbBuyAccount entity) { Map result = new HashMap<>(); if (entity.getTenantId() == null || entity.getTenantId().trim().isEmpty()) { @@ -382,28 +490,55 @@ public class LbBuyAccountServiceImpl /** * 将本次抢单结果与抢单时间写入 lb_buy_account（rush_buy_result、last_rush_buy_time）。 */ - private void persistRushBuyOutcome(String accountId, String resultMessage, LocalDateTime rushBuyTime) { + private void persistRushBuyOutcome(String accountId, String accountLabel, String resultMessage) { if (accountId == null || accountId.trim().isEmpty()) { return; } String storedResult = truncateRushBuyResult( resultMessage != null && !resultMessage.isEmpty() ? resultMessage : "未知抢单结果"); - LocalDateTime time = rushBuyTime != null ? rushBuyTime : LocalDateTime.now(); + LocalDateTime now = LocalDateTime.now(); try { - boolean ok = this.lambdaUpdate() - .eq(LbBuyAccount::getId, accountId.trim()) - .set(LbBuyAccount::getRushBuyResult, storedResult) - .set(LbBuyAccount::getLastRushBuyTime, time) - .set(LbBuyAccount::getUpdateTime, LocalDateTime.now()) - .update(); - if (!ok) { - log.warn("抢单结果落库未更新任何行，accountId={}", accountId); + int rows = this.baseMapper.updateRushBuyOutcome( + accountId.trim(), storedResult, now, now); + if (rows <= 0) { + log.warn("抢单结果落库未更新任何行，{}，accountId={}，lastRushBuyTime={}", + accountLabel, accountId, now); + } else { + log.info("抢单结果已落库，{}，lastRushBuyTime={}，rushBuyResult={}", + accountLabel, now, storedResult); } } catch (Exception e) { - log.error("抢单结果落库异常，accountId={}", accountId, e); + log.error("抢单结果落库异常，{}，accountId={}，lastRushBuyTime={}", + accountLabel, accountId, now, e); } } + private static String formatRushBuyAccountLabel(String accountId, LbBuyAccount account) { + StringBuilder sb = new StringBuilder("抢单账号"); + if (account != null) { + if (account.getLoginAccount() != null && !account.getLoginAccount().trim().isEmpty()) { + sb.append(" loginAccount=").append(account.getLoginAccount().trim()); + } + if (account.getNickname() != null && !account.getNickname().trim().isEmpty()) { + sb.append(" nickname=").append(account.getNickname().trim()); + } + } + if (accountId != null && !accountId.trim().isEmpty()) { + sb.append(" accountId=").append(accountId.trim()); + } + return sb.toString(); + } + + private static String attachAccountLabel(String accountLabel, String message) { + if (accountLabel == null || accountLabel.isEmpty()) { + return message; + } + if (message == null || message.isEmpty()) { + return accountLabel; + } + return accountLabel + " | " + message; + } + private static String formatPreRushBuyFailure(String reason) { return "【失败】原因：" + reason; } diff --git a/src/main/java/com/rj/service/impl/LbGoodsServiceImpl.java b/src/main/java/com/rj/service/impl/LbGoodsServiceImpl.java index 73cdd69..264ce06 100644 --- a/src/main/java/com/rj/service/impl/LbGoodsServiceImpl.java +++ b/src/main/java/com/rj/service/impl/LbGoodsServiceImpl.java @@ -34,7 +34,7 @@ import java.util.Optional; @RequiredArgsConstructor public class LbGoodsServiceImpl extends ServiceImpl implements ILbGoodsService { - private static final BigDecimal RUSH_BUY_MAX_TOTAL_MONEY = new BigDecimal("39000"); + private static final BigDecimal RUSH_BUY_MIN_TOTAL_MONEY = new BigDecimal("29000"); private static final DateTimeFormatter DATETIME_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @@ -375,6 +375,120 @@ public class LbGoodsServiceImpl extends ServiceImpl impl } } + @Override + public Map fetchGoodsFromHxrInMemory(String tenantId, String token) { + Map result = new HashMap<>(); + try { + if (tenantId == null || tenantId.trim().isEmpty()) { + result.put("success", false); + result.put("message", "tenantId不能为空"); + result.put("data", List.of()); + result.put("fetched", 0); + return result; + } + + String tid = tenantId.trim(); + Optional apiContextOpt = + lbThirdIntegrationConfigService.resolveGoodsApiContext(tid, token); + if (apiContextOpt.isEmpty()) { + result.put("success", false); + result.put("message", + "未找到该租户的第三方集成配置，或配置未启用、URL/凭证不完整（请检查 lb_third_integration_config）"); + result.put("data", List.of()); + result.put("fetched", 0); + return result; + } + HxrGoodsApiContext apiContext = apiContextOpt.get(); + + String previousTenantId = TenantContextHolder.getTenantId(); + TenantContextHolder.setTenantId(tid); + try { + HxrLbGoodsPageData firstPageData = null; + int page = 1; + List allGoods = new ArrayList<>(); + while (true) { + Optional opt = + hxrAdminGoodsService.fetchGoodsPage(page, apiContext); + if (opt.isEmpty()) { + if (page == 1) { + result.put("success", false); + result.put("message", + "未拉取到货品（请检查 lb_third_integration_config 中 token、appStr、URL 或网络）"); + result.put("data", List.of()); + result.put("fetched", 0); + return result; + } + result.put("success", false); + result.put("message", "第 " + page + " 页拉取失败，已成功拉取前 " + + (page - 1) + " 页共 " + allGoods.size() + " 条"); + result.put("data", List.copyOf(allGoods)); + result.put("fetched", allGoods.size()); + if (firstPageData != null) { + result.put("lastPage", firstPageData.lastPage()); + } + result.put("pages", page - 1); + return result; + } + + HxrLbGoodsSelectResponse body = opt.get(); + HxrLbGoodsPageData pageData = body.data(); + if (firstPageData == null && pageData != null) { + firstPageData = pageData; + } + List rows = pageData != null ? pageData.list() : null; + if (rows == null || rows.isEmpty()) { + break; + } + + LocalDateTime now = LocalDateTime.now(); + for (LbGoods row : rows) { + if (row.getId() == null) { + continue; + } + row.setTenantId(tid); + row.setCurrentPage(page); + applyDefaults(row); + if (row.getCreatedAt() == null) { + row.setCreatedAt(now); + } + if (row.getUpdatedAt() == null) { + row.setUpdatedAt(now); + } + allGoods.add(row); + } + + boolean hasMore = pageData != null && pageData.hasmore(); + int lastPage = pageData != null ? pageData.lastPage() : page; + if (!hasMore || page >= lastPage) { + break; + } + page++; + } + + result.put("success", true); + result.put("message", allGoods.isEmpty() ? "接口成功，无货品数据" : "拉取完成"); + result.put("data", List.copyOf(allGoods)); + result.put("fetched", allGoods.size()); + result.put("lastPage", firstPageData != null ? firstPageData.lastPage() : page); + result.put("pages", page); + return result; + } finally { + if (previousTenantId != null) { + TenantContextHolder.setTenantId(previousTenantId); + } else { + TenantContextHolder.clear(); + } + } + } catch (Exception e) { + result.put("success", false); + result.put("message", "拉取异常：" + e.getMessage()); + result.put("data", List.of()); + result.put("fetched", 0); + log.error("fetchGoodsFromHxrInMemory failed", e); + return result; + } + } + private int upsertBatch(List entities) { if (entities == null || entities.isEmpty()) { return 0; @@ -399,8 +513,9 @@ public class LbGoodsServiceImpl extends ServiceImpl impl } @Override - public Map rushBuy(String tenantId, String token, Integer maxBuyCount) { + public Map rushBuy(String tenantId, String token, Integer maxBuyCount, String rushBuyAccountLabel) { Map result = new HashMap<>(); + String accountTag = formatRushBuyAccountLogTag(rushBuyAccountLabel); try { if (tenantId == null || tenantId.trim().isEmpty()) { result.put("success", false); @@ -430,22 +545,24 @@ public class LbGoodsServiceImpl extends ServiceImpl impl String previousTenantId = TenantContextHolder.getTenantId(); TenantContextHolder.setTenantId(tid); try { + LocalDateTime updatedAfter = LocalDateTime.now().minusDays(1); LambdaQueryWrapper w = new LambdaQueryWrapper<>(); w.eq(LbGoods::getTenantId, tid); w.isNotNull(LbGoods::getSellerId); - w.lt(LbGoods::getTotalMoney, RUSH_BUY_MAX_TOTAL_MONEY); - w.orderByDesc(LbGoods::getTotalMoney).orderByDesc(LbGoods::getId); + w.gt(LbGoods::getTotalMoney, RUSH_BUY_MIN_TOTAL_MONEY); + w.ge(LbGoods::getUpdatedAt, updatedAfter); + w.orderByDesc(LbGoods::getUpdatedAt); List goodsList = this.list(w); if (goodsList.isEmpty()) { result.put("success", false); - result.put("message", "lb_goods 中无可抢购货品（金额需小于39000）"); + result.put("message", "lb_goods 中无可抢购货品（金额需大于29000，且更新时间需在24小时内）"); result.put("successCount", 0); result.put("failCount", 0); result.put("details", List.of()); return result; } - int maxAttempts = maxBuyCount * 2; + int maxAttempts = maxBuyCount * 100; List> details = new ArrayList<>(); int successCount = 0; int failCount = 0; @@ -472,7 +589,11 @@ public class LbGoodsServiceImpl extends ServiceImpl impl item.put("sellerId", goods.getSellerId()); item.put("title", goods.getTitle()); item.put("totalMoney", goods.getTotalMoney()); - log.info("正在抢购货品是：{}",item.toString()); + if (accountTag.isEmpty()) { + log.info("正在抢购货品：{}", item); + } else { + log.info("{} 正在抢购货品：{}", accountTag, item); + } HxrAdminBuyService.BuyApiResult buyResult = hxrAdminBuyService.buy(goods.getId(), goods.getSellerId(), apiContext); item.put("apiCode", buyResult.apiCode()); @@ -498,6 +619,11 @@ public class LbGoodsServiceImpl extends ServiceImpl impl result.put("stoppedByMax", stoppedByMax); result.put("stoppedByMaxAttempts", stoppedByMaxAttempts); result.put("details", details); + if (accountTag.isEmpty()) { + log.info("抢购结束，success={}，message={}", successCount > 0, result.get("message")); + } else { + log.info("{} 抢购结束，success={}，message={}", accountTag, successCount > 0, result.get("message")); + } return result; } finally { if (previousTenantId != null) { @@ -507,13 +633,24 @@ public class LbGoodsServiceImpl extends ServiceImpl impl } } } catch (Exception e) { - log.error("rushBuy failed", e); + if (accountTag.isEmpty()) { + log.error("rushBuy failed", e); + } else { + log.error("{} rushBuy failed", accountTag, e); + } result.put("success", false); result.put("message", "抢购异常：" + e.getMessage()); return result; } } + private static String formatRushBuyAccountLogTag(String rushBuyAccountLabel) { + if (rushBuyAccountLabel == null || rushBuyAccountLabel.trim().isEmpty()) { + return ""; + } + return rushBuyAccountLabel.trim(); + } + private static void applyDefaults(LbGoods entity) { if (entity.getPrice() == null) { entity.setPrice(BigDecimal.ZERO);