From 78c1fd60a7cc7c626abf8129b63ca77aeffceed9 Mon Sep 17 00:00:00 2001 From: cst61 Date: Fri, 5 Jun 2026 10:12:29 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84=E6=8A=A2=E5=8D=95=E4=BB=A3?= =?UTF-8?q?=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../rj/controller/LbBuyAccountController.java | 1 + .../com/rj/controller/LbGoodsController.java | 2 +- .../rj/dto/LbBuyAccountRushBuyContext.java | 10 + .../com/rj/dto/LbRushBuyGoodsCoordinator.java | 74 +++++++ .../com/rj/mapper/LbBuyAccountMapper.java | 14 ++ .../com/rj/service/HxrAdminBuyService.java | 32 +++ .../rj/service/HxrAdminUserLoginService.java | 30 +++ .../java/com/rj/service/ILbGoodsService.java | 33 +++- .../rj/service/LbBuyAccountTokenService.java | 97 +++++++++ .../service/impl/LbBuyAccountServiceImpl.java | 187 +++++++++++------- .../impl/LbFanManagementServiceImpl.java | 18 +- .../rj/service/impl/LbGoodsServiceImpl.java | 156 ++++++++++++++- 12 files changed, 552 insertions(+), 102 deletions(-) create mode 100644 src/main/java/com/rj/dto/LbBuyAccountRushBuyContext.java create mode 100644 src/main/java/com/rj/dto/LbRushBuyGoodsCoordinator.java create mode 100644 src/main/java/com/rj/service/LbBuyAccountTokenService.java diff --git a/src/main/java/com/rj/controller/LbBuyAccountController.java b/src/main/java/com/rj/controller/LbBuyAccountController.java index 6dab369..60b4223 100644 --- a/src/main/java/com/rj/controller/LbBuyAccountController.java +++ b/src/main/java/com/rj/controller/LbBuyAccountController.java @@ -88,6 +88,7 @@ public class LbBuyAccountController { summary = "按抢单账号批量抢购", description = "根据 ids 查询 lb_buy_account，按 tenant_id 关联 lb_third_integration_config 获取 URL 与 appStr；" + + "各账号并发抢购（每账号一线程）；accountResults 顺序与 ids 一致；同租户同货品 ID 仅允许一个账号尝试；" + "每个账号使用 token_front 作为 hxrd 请求 Token，max_grab_count 作为最大成功抢购笔数，" + "抢购逻辑同 LbGoodsController#rushBuy") public ResponseEntity> rushBuy( diff --git a/src/main/java/com/rj/controller/LbGoodsController.java b/src/main/java/com/rj/controller/LbGoodsController.java index bf48ce1..a352ceb 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 选取金额大于29000且24小时内更新的货品，调用 hxrd POST /api/order/buy（body: id、seller_id）；" + "从 lb_goods 选取金额大于25000且不超过38000、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/dto/LbBuyAccountRushBuyContext.java b/src/main/java/com/rj/dto/LbBuyAccountRushBuyContext.java new file mode 100644 index 0000000..8857a76 --- /dev/null +++ b/src/main/java/com/rj/dto/LbBuyAccountRushBuyContext.java @@ -0,0 +1,10 @@ +package com.rj.dto; + +/** + * 抢单时 token 失效后自动登录刷新 {@code lb_buy_account.token_front} 所需账号信息。 + */ +public record LbBuyAccountRushBuyContext( + String accountId, + String loginAccount, + String loginPassword) { +} diff --git a/src/main/java/com/rj/dto/LbRushBuyGoodsCoordinator.java b/src/main/java/com/rj/dto/LbRushBuyGoodsCoordinator.java new file mode 100644 index 0000000..96e4bc6 --- /dev/null +++ b/src/main/java/com/rj/dto/LbRushBuyGoodsCoordinator.java @@ -0,0 +1,74 @@ +package com.rj.dto; + +import java.util.concurrent.ConcurrentHashMap; + +/** + * 单次批量抢单会话内，保证同一租户的同一货品 ID 最多被一个账号尝试抢购。 + */ +public class LbRushBuyGoodsCoordinator { + + /** 货品已被外部系统抢订，本批次内任何账号均不可再抢。 */ + private static final String UNAVAILABLE = "__UNAVAILABLE__"; + + private final ConcurrentHashMap claimedBy = new ConcurrentHashMap<>(); + + /** 任一线程收到「活动未开始」后置位，本批次内其他账号不再发起抢购。 */ + private volatile boolean activityNotStarted; + + private static String key(String tenantId, Object goodsId) { + return tenantId.trim() + ":" + goodsId; + } + + /** + * @return {@code true} 表示当前账号取得该货品的独占权；{@code false} 表示已被其他账号占用 + */ + public boolean tryAcquire(String tenantId, Object goodsId, String accountId) { + if (tenantId == null || tenantId.trim().isEmpty() + || goodsId == null || accountId == null || accountId.trim().isEmpty()) { + return true; + } + String k = key(tenantId, goodsId); + String existing = claimedBy.get(k); + if (UNAVAILABLE.equals(existing)) { + return false; + } + String holder = claimedBy.putIfAbsent(k, accountId.trim()); + return holder == null || holder.equals(accountId.trim()); + } + + /** + * 接口返回「此订单已被抢订」时调用，本批次内同租户该货品不再分配给任何账号。 + */ + public void markUnavailable(String tenantId, Object goodsId) { + if (tenantId == null || tenantId.trim().isEmpty() || goodsId == null) { + return; + } + claimedBy.put(key(tenantId, goodsId), UNAVAILABLE); + } + + public boolean isUnavailable(String tenantId, Object goodsId) { + if (tenantId == null || tenantId.trim().isEmpty() || goodsId == null) { + return false; + } + return UNAVAILABLE.equals(claimedBy.get(key(tenantId, goodsId))); + } + + /** + * 抢购失败时释放独占权，便于后续账号继续尝试同一货品。 + */ + public void markActivityNotStarted() { + activityNotStarted = true; + } + + public boolean isActivityNotStarted() { + return activityNotStarted; + } + + public void release(String tenantId, Object goodsId, String accountId) { + if (tenantId == null || tenantId.trim().isEmpty() + || goodsId == null || accountId == null || accountId.trim().isEmpty()) { + return; + } + claimedBy.remove(key(tenantId, goodsId), accountId.trim()); + } +} diff --git a/src/main/java/com/rj/mapper/LbBuyAccountMapper.java b/src/main/java/com/rj/mapper/LbBuyAccountMapper.java index 821f74f..5ded313 100644 --- a/src/main/java/com/rj/mapper/LbBuyAccountMapper.java +++ b/src/main/java/com/rj/mapper/LbBuyAccountMapper.java @@ -43,4 +43,18 @@ public interface LbBuyAccountMapper extends BaseMapper { @Param("rushBuyResult") String rushBuyResult, @Param("lastRushBuyTime") LocalDateTime lastRushBuyTime, @Param("updateTime") LocalDateTime updateTime); + + /** + * 按主键更新 token_front；跳过多租户拦截，避免批量抢单 worker 线程无租户上下文导致更新 0 行。 + */ + @InterceptorIgnore(tenantLine = "true") + @Update(""" + UPDATE lb_buy_account + SET token_front = #{tokenFront}, + update_time = #{updateTime} + WHERE id = #{id} + """) + int updateTokenFront(@Param("id") String id, + @Param("tokenFront") String tokenFront, + @Param("updateTime") LocalDateTime updateTime); } diff --git a/src/main/java/com/rj/service/HxrAdminBuyService.java b/src/main/java/com/rj/service/HxrAdminBuyService.java index 8206cd6..e94f24b 100644 --- a/src/main/java/com/rj/service/HxrAdminBuyService.java +++ b/src/main/java/com/rj/service/HxrAdminBuyService.java @@ -40,6 +40,15 @@ public class HxrAdminBuyService { /** 外部系统返回此文案时表示该账号当日可抢订单数已满，应停止继续抢购。 */ public static final String MSG_DAILY_LIMIT_EXCEEDED = "已超出当天可抢订单数"; + /** 外部系统返回此文案或 {@code code=401} 时表示 token 失效或未登录，应停止继续抢购。 */ + public static final String MSG_LOGIN_REQUIRED = "请登录"; + + /** 外部系统返回此文案时表示抢购活动尚未开始。 */ + public static final String MSG_ACTIVITY_NOT_STARTED = "活动未开始"; + + /** 外部系统返回此文案时表示该货品已被他人抢订，同批次其他账号不应再尝试。 */ + public static final String MSG_ORDER_ALREADY_GRABBED = "此订单已被抢订"; + private static final ObjectMapper JSON = new ObjectMapper() .setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE); @@ -151,10 +160,18 @@ public class HxrAdminBuyService { log.warn("hxr /api/order/buy 鉴权失败常见原因：Token 与浏览器不一致或已过期、goodsApiAppStr 错误；" + "请在 rush-buy 请求体传入浏览器 DevTools 中 Token 头的值，并核对 lb_third_integration_config.goods_api_token / goods_api_app_str"); } + if (msg != null && msg.contains(MSG_ACTIVITY_NOT_STARTED)) { + log.warn("hxr /api/order/buy 活动未开始，调用方应停止继续抢购 id={} sellerId={}", + goodsId, sellerId); + } if (MSG_DAILY_LIMIT_EXCEEDED.equals(msg)) { log.warn("hxr /api/order/buy 当日可抢订单数已满，停止该账号继续抢购 id={} sellerId={} tokenPrefix={}", goodsId, sellerId, abbreviate(resolvedToken, 8)); } + if (code == 401 || MSG_LOGIN_REQUIRED.equals(msg)) { + log.warn("hxr /api/order/buy token 失效或未登录，停止该账号继续抢购 id={} sellerId={} tokenPrefix={}", + goodsId, sellerId, abbreviate(resolvedToken, 8)); + } } return new BuyApiResult(code == 0, code, msg); } @@ -181,6 +198,21 @@ public class HxrAdminBuyService { public boolean dailyLimitExceeded() { return apiCode == 1 && MSG_DAILY_LIMIT_EXCEEDED.equals(apiMsg); } + + /** {@code code=401} 或 msg 为「请登录」时，调用方应停止该账号后续抢购。 */ + public boolean loginRequired() { + return apiCode == 401 || MSG_LOGIN_REQUIRED.equals(apiMsg); + } + + /** msg 包含「此订单已被抢订」时，同租户该货品在批量抢单中应标记为不可再抢。 */ + public boolean orderAlreadyGrabbed() { + return apiMsg != null && apiMsg.contains(MSG_ORDER_ALREADY_GRABBED); + } + + /** msg 包含「活动未开始」时，调用方应立刻停止该账号及本批次后续抢购。 */ + public boolean activityNotStarted() { + return apiMsg != null && apiMsg.contains(MSG_ACTIVITY_NOT_STARTED); + } } private static String randomNoncestr() { diff --git a/src/main/java/com/rj/service/HxrAdminUserLoginService.java b/src/main/java/com/rj/service/HxrAdminUserLoginService.java index c48d990..8fc4846 100644 --- a/src/main/java/com/rj/service/HxrAdminUserLoginService.java +++ b/src/main/java/com/rj/service/HxrAdminUserLoginService.java @@ -120,6 +120,36 @@ public class HxrAdminUserLoginService { return new LoginApiResult(ok, code, msg, parsed, httpStatus); } + @SuppressWarnings("unchecked") + public static Map extractUserinfo(Object parsed) { + if (!(parsed instanceof Map root)) { + return null; + } + Object dataObj = root.get("data"); + if (!(dataObj instanceof Map data)) { + return null; + } + Object userinfoObj = data.get("userinfo"); + if (!(userinfoObj instanceof Map)) { + return null; + } + return (Map) userinfoObj; + } + + /** 从登录 API 解析结果中提取 {@code data.userinfo.token}。 */ + public static String extractToken(Object parsed) { + Map userinfo = extractUserinfo(parsed); + if (userinfo == null) { + return null; + } + Object tokenObj = userinfo.get("token"); + if (tokenObj == null) { + return null; + } + String token = tokenObj.toString().trim(); + return token.isEmpty() ? null : token; + } + public record LoginApiResult( boolean success, int apiCode, diff --git a/src/main/java/com/rj/service/ILbGoodsService.java b/src/main/java/com/rj/service/ILbGoodsService.java index d602be6..7ddbb6d 100644 --- a/src/main/java/com/rj/service/ILbGoodsService.java +++ b/src/main/java/com/rj/service/ILbGoodsService.java @@ -1,6 +1,8 @@ package com.rj.service; import com.baomidou.mybatisplus.extension.service.IService; +import com.rj.dto.LbBuyAccountRushBuyContext; +import com.rj.dto.LbRushBuyGoodsCoordinator; import com.rj.entity.LbGoods; import java.util.Map; @@ -44,15 +46,38 @@ public interface ILbGoodsService extends IService { Map fetchGoodsFromHxrInMemory(String tenantId, String token); /** - * 选取金额大于29000且24小时内更新的 {@code lb_goods}，调用 hxrd {@code /api/order/buy} 抢购； + * 选取金额大于25000且不超过38000、24小时内更新的 {@code lb_goods}，调用 hxrd {@code /api/order/buy} 抢购； * 成功笔数达到 {@code maxBuyCount} 后停止，且循环抢购总次数不超过 {@code maxBuyCount} 的5倍。 * * @param token 可选；非空时作为请求头 token，否则使用 {@code lb_third_integration_config} 中的 goodsApiToken - * @param ；批量账号抢单时传入，用于日志标识当前抢单账号 + * @param rushBuyAccountLabel 批量账号抢单时传入，用于日志标识当前抢单账号 + * @param tokenRefreshContext 非空且 token 失效时，按账号凭证自动登录并刷新 {@code lb_buy_account.token_front} + * @param goodsCoordinator 批量抢单时传入，保证同租户同货品 ID 仅一个账号抢购 + * @param accountId 批量抢单时当前账号主键，与 {@code goodsCoordinator} 配合使用 */ default Map rushBuy(String tenantId, String token, Integer maxBuyCount) { - return rushBuy(tenantId, token, maxBuyCount, null); + return rushBuy(tenantId, token, maxBuyCount, null, null, null, null); } - Map rushBuy(String tenantId, String token, Integer maxBuyCount, String rushBuyAccountLabel); + default Map rushBuy(String tenantId, String token, Integer maxBuyCount, String rushBuyAccountLabel) { + return rushBuy(tenantId, token, maxBuyCount, rushBuyAccountLabel, null, null, null); + } + + default Map rushBuy( + String tenantId, + String token, + Integer maxBuyCount, + String rushBuyAccountLabel, + LbBuyAccountRushBuyContext tokenRefreshContext) { + return rushBuy(tenantId, token, maxBuyCount, rushBuyAccountLabel, tokenRefreshContext, null, null); + } + + Map rushBuy( + String tenantId, + String token, + Integer maxBuyCount, + String rushBuyAccountLabel, + LbBuyAccountRushBuyContext tokenRefreshContext, + LbRushBuyGoodsCoordinator goodsCoordinator, + String accountId); } diff --git a/src/main/java/com/rj/service/LbBuyAccountTokenService.java b/src/main/java/com/rj/service/LbBuyAccountTokenService.java new file mode 100644 index 0000000..fc10cf3 --- /dev/null +++ b/src/main/java/com/rj/service/LbBuyAccountTokenService.java @@ -0,0 +1,97 @@ +package com.rj.service; + +import com.rj.dto.hxr.HxrUserLoginApiContext; +import com.rj.mapper.LbBuyAccountMapper; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.time.LocalDateTime; +import java.util.Optional; + +/** + * 抢单账号 token 刷新：参考 {@link com.rj.service.impl.LbFanManagementServiceImpl#simulateLogin} 调用登录 API， + * 从 {@code data.userinfo.token} 解析并写入 {@code lb_buy_account.token_front}。 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class LbBuyAccountTokenService { + + private final ILbThirdIntegrationConfigService lbThirdIntegrationConfigService; + private final HxrAdminUserLoginService hxrAdminUserLoginService; + private final LbBuyAccountMapper lbBuyAccountMapper; + + /** + * @return 新 token；登录失败、响应无 token 或落库失败时为空 + */ + public Optional loginAndPersistTokenFront( + String accountId, + String tenantId, + String loginAccount, + String loginPassword) { + if (accountId == null || accountId.isBlank()) { + log.warn("自动登录刷新 token 失败：accountId 为空"); + return Optional.empty(); + } + if (tenantId == null || tenantId.isBlank()) { + log.warn("自动登录刷新 token 失败：accountId={} tenantId 为空", accountId); + return Optional.empty(); + } + if (loginAccount == null || loginAccount.isBlank()) { + log.warn("自动登录刷新 token 失败：accountId={} login_account 为空", accountId); + return Optional.empty(); + } + if (loginPassword == null || loginPassword.isBlank()) { + log.warn("自动登录刷新 token 失败：accountId={} login_password 为空", accountId); + return Optional.empty(); + } + + Optional ctxOpt = + lbThirdIntegrationConfigService.resolveUserLoginApiContext(tenantId.trim()); + if (ctxOpt.isEmpty()) { + log.warn("自动登录刷新 token 失败：accountId={} 未找到租户登录 API 配置", accountId); + return Optional.empty(); + } + + try { + HxrAdminUserLoginService.LoginApiResult loginResult = hxrAdminUserLoginService.login( + loginAccount.trim(), + loginPassword.trim(), + ctxOpt.get()); + if (!loginResult.success()) { + log.warn("自动登录刷新 token 失败：accountId={} apiCode={} apiMsg={}", + accountId, loginResult.apiCode(), loginResult.apiMsg()); + return Optional.empty(); + } + + String token = HxrAdminUserLoginService.extractToken(loginResult.parsed()); + if (token == null || token.isBlank()) { + log.warn("自动登录刷新 token 失败：accountId={} 响应中无 data.userinfo.token", accountId); + return Optional.empty(); + } + + String trimmedToken = token.trim(); + LocalDateTime now = LocalDateTime.now(); + int updated = lbBuyAccountMapper.updateTokenFront(accountId.trim(), trimmedToken, now); + if (updated <= 0) { + log.warn("自动登录 token 解析成功但落库失败：accountId={}", accountId); + return Optional.empty(); + } + + log.info("自动登录成功，token_front 已更新：accountId={} tokenPrefix={}", + accountId, abbreviate(trimmedToken, 8)); + return Optional.of(trimmedToken); + } catch (Exception e) { + log.error("自动登录刷新 token 异常：accountId={}", accountId, e); + return Optional.empty(); + } + } + + private static String abbreviate(String s, int maxLen) { + if (s == null) { + return ""; + } + return s.length() <= maxLen ? s : s.substring(0, maxLen) + "..."; + } +} diff --git a/src/main/java/com/rj/service/impl/LbBuyAccountServiceImpl.java b/src/main/java/com/rj/service/impl/LbBuyAccountServiceImpl.java index 52b0e1a..3082f7c 100644 --- a/src/main/java/com/rj/service/impl/LbBuyAccountServiceImpl.java +++ b/src/main/java/com/rj/service/impl/LbBuyAccountServiceImpl.java @@ -3,6 +3,8 @@ package com.rj.service.impl; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; +import com.rj.dto.LbBuyAccountRushBuyContext; +import com.rj.dto.LbRushBuyGoodsCoordinator; import com.rj.entity.LbBuyAccount; import com.rj.mapper.LbBuyAccountMapper; import com.rj.service.ILbBuyAccountService; @@ -19,7 +21,7 @@ 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.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -34,6 +36,9 @@ public class LbBuyAccountServiceImpl /** 与表字段 rush_buy_result（TEXT）长度上限一致 */ private static final int RUSH_BUY_RESULT_MAX_LEN = 65535; + /** 单次批量抢单最多并发线程数（与账号数取 min，每账号一线程） */ + private static final int RUSH_BUY_MAX_PARALLEL_THREADS = 64; + private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @@ -231,68 +236,35 @@ public class LbBuyAccountServiceImpl } } - int accountCount = normalizedIds.size(); - AtomicInteger threadSeq = new AtomicInteger(0); - ExecutorService batchExecutor = Executors.newFixedThreadPool(accountCount, r -> { + LbRushBuyGoodsCoordinator goodsCoordinator = new LbRushBuyGoodsCoordinator(); + int poolSize = Math.min(normalizedIds.size(), RUSH_BUY_MAX_PARALLEL_THREADS); + AtomicInteger threadSeq = new AtomicInteger(); + ExecutorService executor = Executors.newFixedThreadPool(poolSize, r -> { Thread t = new Thread(r, "lb-rush-buy-" + threadSeq.incrementAndGet()); - t.setDaemon(true); + t.setDaemon(false); return t; }); - log.info("批量抢单启动，账号数={}，独立线程数={}", accountCount, accountCount); + log.info("批量抢单启动，账号数={}，并发线程数={}（每账号一线程），同租户同货品仅一个账号抢购", + normalizedIds.size(), poolSize); + List> accountResults; + int successAccounts; + int failAccounts; try { - CountDownLatch startLatch = new CountDownLatch(1); - CountDownLatch readyLatch = new CountDownLatch(accountCount); - CountDownLatch doneLatch = new CountDownLatch(accountCount); - @SuppressWarnings("unchecked") - Map[] resultArray = new Map[accountCount]; - - for (int i = 0; i < accountCount; i++) { - final int index = i; - String id = normalizedIds.get(i); + List>> futures = new ArrayList<>(normalizedIds.size()); + for (String id : normalizedIds) { 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(); - } - }); + futures.add(CompletableFuture.supplyAsync( + () -> runSingleAccountRushBuyTask(id, account, goodsCoordinator), + executor)); } + CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join(); - if (!readyLatch.await(30, TimeUnit.SECONDS)) { - result.put("success", false); - result.put("message", "抢单线程就绪超时"); - return result; - } - log.info("全部 {} 个抢单线程已就绪，同时启动", accountCount); - startLatch.countDown(); - - if (!doneLatch.await(120, TimeUnit.SECONDS)) { - result.put("success", false); - result.put("message", "批量抢单执行超时"); - return result; - } - - List> accountResults = new ArrayList<>(accountCount); - int successAccounts = 0; - int failAccounts = 0; - for (Map item : resultArray) { + accountResults = new ArrayList<>(normalizedIds.size()); + successAccounts = 0; + failAccounts = 0; + for (CompletableFuture> future : futures) { + Map item = future.join(); accountResults.add(item); if (Boolean.TRUE.equals(item.get("success"))) { successAccounts++; @@ -300,27 +272,29 @@ public class LbBuyAccountServiceImpl 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(); + executor.shutdown(); try { - if (!batchExecutor.awaitTermination(120, TimeUnit.SECONDS)) { - batchExecutor.shutdownNow(); + if (!executor.awaitTermination(2, TimeUnit.HOURS)) { + log.warn("批量抢单线程池未在 2 小时内结束，执行 shutdownNow"); + executor.shutdownNow(); } } catch (InterruptedException e) { - batchExecutor.shutdownNow(); + executor.shutdownNow(); Thread.currentThread().interrupt(); + throw new RuntimeException("批量抢单等待线程结束被中断", e); } } + + 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); @@ -329,6 +303,20 @@ public class LbBuyAccountServiceImpl } } + private Map runSingleAccountRushBuyTask( + String id, LbBuyAccount account, LbRushBuyGoodsCoordinator goodsCoordinator) { + try { + return processSingleAccountRushBuy(id, account, goodsCoordinator); + } catch (Exception e) { + log.error("抢单账号任务执行异常，accountId={}，loginAccount={}，thread={}", + id, account != null ? account.getLoginAccount() : null, + Thread.currentThread().getName(), e); + Map item = buildRushBuyTaskErrorItem(id, account, e.getMessage()); + persistRushBuyOutcomeFromItem(account, item); + return item; + } + } + private static Map buildRushBuyTaskErrorItem(String accountId, LbBuyAccount account, String reason) { String accountLabel = formatRushBuyAccountLabel(accountId, account); Map item = new LinkedHashMap<>(); @@ -344,9 +332,10 @@ public class LbBuyAccountServiceImpl } /** - * 单账号抢单逻辑，供线程池并行调用。 + * 单账号抢单逻辑；批量入口为每账号一线程并发调用，共享 {@link LbRushBuyGoodsCoordinator}。 */ - private Map processSingleAccountRushBuy(String id, LbBuyAccount account) { + private Map processSingleAccountRushBuy( + String id, LbBuyAccount account, LbRushBuyGoodsCoordinator goodsCoordinator) { String accountLabel = formatRushBuyAccountLabel(id, account); log.info("抢单任务开始，{}，thread={}", accountLabel, Thread.currentThread().getName()); Map item = new LinkedHashMap<>(); @@ -378,24 +367,41 @@ public class LbBuyAccountServiceImpl if (maxGrabCount == null || maxGrabCount <= 0) { resultMessage = attachAccountLabel(accountLabel, formatPreRushBuyFailure("maxGrabCount必须大于0")); } else { + LbBuyAccountRushBuyContext tokenRefreshContext = null; + if (account.getLoginAccount() != null && !account.getLoginAccount().trim().isEmpty() + && account.getLoginPassword() != null && !account.getLoginPassword().trim().isEmpty()) { + tokenRefreshContext = new LbBuyAccountRushBuyContext( + account.getId(), + account.getLoginAccount().trim(), + account.getLoginPassword().trim()); + } Map rushBuyResult = lbGoodsService.rushBuy( account.getTenantId().trim(), account.getTokenFront().trim(), maxGrabCount, - accountLabel); + accountLabel, + tokenRefreshContext, + goodsCoordinator, + account.getId()); item.put("rushBuyResult", rushBuyResult); rushSuccess = rushBuyResult.get("success") instanceof Boolean ? (Boolean) rushBuyResult.get("success") : null; resultMessage = attachAccountLabel( accountLabel, formatRushBuyResultMessage(rushBuyResult)); + persistRushBuyOutcome( + account.getId(), + accountLabel, + formatRushBuyResultForDb(accountLabel, 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); + if (account != null && account.getId() != null && item.get("rushBuyResult") == null) { + persistRushBuyOutcome(account.getId(), accountLabel, resultMessage); + } return item; } @@ -574,7 +580,7 @@ public class LbBuyAccountServiceImpl StringBuilder sb = new StringBuilder(); if (summaryMsg != null) { - sb.append("汇总：").append(summaryMsg); + sb.append(summaryMsg); } Object successCount = rushBuyResult.get("successCount"); Object failCount = rushBuyResult.get("failCount"); @@ -597,6 +603,39 @@ public class LbBuyAccountServiceImpl return truncateRushBuyResult(sb.toString()); } + /** + * 落库文案：将 {@code rushBuyResult.message} 置于最前，便于在库中快速识别本次抢购结论。 + */ + private static String formatRushBuyResultForDb(String accountLabel, Map rushBuyResult) { + if (rushBuyResult == null || rushBuyResult.isEmpty()) { + return truncateRushBuyResult(formatPreRushBuyFailure("抢购未返回结果")); + } + + Object summaryMsg = rushBuyResult.get("message"); + String summary = summaryMsg != null ? summaryMsg.toString().trim() : ""; + String formattedBody = formatRushBuyResultMessage(rushBuyResult); + + Object detailsObj = rushBuyResult.get("details"); + if (!(detailsObj instanceof List details) || details.isEmpty()) { + if (summary.isEmpty()) { + return truncateRushBuyResult(attachAccountLabel(accountLabel, formattedBody)); + } + return truncateRushBuyResult(summary + " | " + accountLabel); + } + + if (summary.isEmpty()) { + return truncateRushBuyResult(attachAccountLabel(accountLabel, formattedBody)); + } + + int sep = formattedBody.indexOf(" | "); + if (sep < 0) { + return truncateRushBuyResult(summary + " | " + accountLabel); + } + String head = formattedBody.substring(0, sep); + String tail = formattedBody.substring(sep + 3); + return truncateRushBuyResult(head + " | " + accountLabel + " | " + tail); + } + /** rushBuy 未产生逐笔明细（配置缺失、无可抢货品、异常等） */ private static String formatRushBuyWithoutDetails(Map rushBuyResult, Object summaryMsg) { Boolean overallSuccess = rushBuyResult.get("success") instanceof Boolean diff --git a/src/main/java/com/rj/service/impl/LbFanManagementServiceImpl.java b/src/main/java/com/rj/service/impl/LbFanManagementServiceImpl.java index f42a64c..c82faa1 100644 --- a/src/main/java/com/rj/service/impl/LbFanManagementServiceImpl.java +++ b/src/main/java/com/rj/service/impl/LbFanManagementServiceImpl.java @@ -305,7 +305,7 @@ public class LbFanManagementServiceImpl item.put("parsed", loginResult.parsed()); if (loginResult.success()) { successCount++; - Map userinfo = extractUserinfo(loginResult.parsed()); + Map userinfo = HxrAdminUserLoginService.extractUserinfo(loginResult.parsed()); if (userinfo != null) { Map saveResult = saveFanFromLoginUserinfo(userinfo, tenantId); @@ -438,22 +438,6 @@ public class LbFanManagementServiceImpl } } - @SuppressWarnings("unchecked") - private static Map extractUserinfo(Object parsed) { - if (!(parsed instanceof Map root)) { - return null; - } - Object dataObj = root.get("data"); - if (!(dataObj instanceof Map data)) { - return null; - } - Object userinfoObj = data.get("userinfo"); - if (!(userinfoObj instanceof Map)) { - return null; - } - return (Map) userinfoObj; - } - private Map saveFanFromLoginUserinfo(Map userinfo, String tenantId) { Map saveResult = new LinkedHashMap<>(); Long fanId = parseLong(userinfo.get("id")); diff --git a/src/main/java/com/rj/service/impl/LbGoodsServiceImpl.java b/src/main/java/com/rj/service/impl/LbGoodsServiceImpl.java index 084b3bb..a56c3f2 100644 --- a/src/main/java/com/rj/service/impl/LbGoodsServiceImpl.java +++ b/src/main/java/com/rj/service/impl/LbGoodsServiceImpl.java @@ -6,12 +6,15 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.rj.dto.hxr.HxrGoodsApiContext; import com.rj.dto.hxr.HxrLbGoodsPageData; import com.rj.dto.hxr.HxrLbGoodsSelectResponse; +import com.rj.dto.LbBuyAccountRushBuyContext; +import com.rj.dto.LbRushBuyGoodsCoordinator; import com.rj.entity.LbGoods; import com.rj.mapper.LbGoodsMapper; import com.rj.service.HxrAdminBuyService; import com.rj.service.HxrAdminGoodsService; import com.rj.service.ILbGoodsService; import com.rj.service.ILbThirdIntegrationConfigService; +import com.rj.service.LbBuyAccountTokenService; import com.rj.tenant.TenantContextHolder; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -35,6 +38,7 @@ import java.util.Optional; public class LbGoodsServiceImpl extends ServiceImpl implements ILbGoodsService { private static final BigDecimal RUSH_BUY_MIN_TOTAL_MONEY = new BigDecimal("25000"); + private static final BigDecimal RUSH_BUY_MAX_TOTAL_MONEY = new BigDecimal("38000"); private static final DateTimeFormatter DATETIME_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @@ -44,6 +48,8 @@ public class LbGoodsServiceImpl extends ServiceImpl impl private final ILbThirdIntegrationConfigService lbThirdIntegrationConfigService; + private final LbBuyAccountTokenService lbBuyAccountTokenService; + @Override public Map add(LbGoods entity) { Map result = new HashMap<>(); @@ -513,7 +519,14 @@ public class LbGoodsServiceImpl extends ServiceImpl impl } @Override - public Map rushBuy(String tenantId, String token, Integer maxBuyCount, String rushBuyAccountLabel) { + public Map rushBuy( + String tenantId, + String token, + Integer maxBuyCount, + String rushBuyAccountLabel, + LbBuyAccountRushBuyContext tokenRefreshContext, + LbRushBuyGoodsCoordinator goodsCoordinator, + String accountId) { Map result = new HashMap<>(); String accountTag = formatRushBuyAccountLogTag(rushBuyAccountLabel); try { @@ -550,28 +563,41 @@ public class LbGoodsServiceImpl extends ServiceImpl impl w.eq(LbGoods::getTenantId, tid); w.isNotNull(LbGoods::getSellerId); w.gt(LbGoods::getTotalMoney, RUSH_BUY_MIN_TOTAL_MONEY); + w.le(LbGoods::getTotalMoney, RUSH_BUY_MAX_TOTAL_MONEY); w.ge(LbGoods::getUpdatedAt, updatedAfter); w.orderByDesc(LbGoods::getTotalMoney); List goodsList = this.list(w); if (goodsList.isEmpty()) { result.put("success", false); - result.put("message", "lb_goods 中无可抢购货品（金额需大于29000，且更新时间需在24小时内）"); + result.put("message", "lb_goods 中无可抢购货品（金额需大于25000且不超过38000，且更新时间需在24小时内）"); result.put("successCount", 0); result.put("failCount", 0); result.put("details", List.of()); return result; } - int maxAttempts = maxBuyCount * 100; + int maxAttempts = maxBuyCount * 150; List> details = new ArrayList<>(); int successCount = 0; int failCount = 0; + int skippedCount = 0; int attemptCount = 0; boolean stoppedByMax = false; boolean stoppedByMaxAttempts = false; boolean stoppedByDailyLimit = false; + boolean stoppedByLoginRequired = false; + boolean stoppedByActivityNotStarted = false; + boolean tokenRefreshAttempted = false; + boolean tokenRefreshed = false; + boolean restartAfterTokenRefresh; + do { + restartAfterTokenRefresh = false; for (LbGoods goods : goodsList) { + if (goodsCoordinator != null && goodsCoordinator.isActivityNotStarted()) { + stoppedByActivityNotStarted = true; + break; + } if (successCount >= maxBuyCount) { stoppedByMax = true; break; @@ -584,6 +610,29 @@ public class LbGoodsServiceImpl extends ServiceImpl impl continue; } + if (goodsCoordinator != null && accountId != null + && !goodsCoordinator.tryAcquire(tid, goods.getId(), accountId)) { + skippedCount++; + Map skippedItem = new LinkedHashMap<>(); + skippedItem.put("id", goods.getId()); + skippedItem.put("sellerId", goods.getSellerId()); + skippedItem.put("title", goods.getTitle()); + skippedItem.put("totalMoney", goods.getTotalMoney()); + skippedItem.put("skipped", true); + skippedItem.put("success", false); + String skipReason = goodsCoordinator.isUnavailable(tid, goods.getId()) + ? "该货品已被抢订，不可再抢，我自动跳过" + : "该货品已被其他账号占用，我自动跳过"; + skippedItem.put("message", skipReason); + details.add(skippedItem); + if (accountTag.isEmpty()) { + log.info("跳过货品 {}，{}", goods.getId(), skipReason); + } else { + log.info("{} 跳过货品 {}，{}", accountTag, goods.getId(), skipReason); + } + continue; + } + attemptCount++; Map item = new LinkedHashMap<>(); item.put("id", goods.getId()); @@ -604,8 +653,22 @@ public class LbGoodsServiceImpl extends ServiceImpl impl if (buyResult.success()) { successCount++; + Thread.sleep(500); } else { failCount++; + if (buyResult.orderAlreadyGrabbed()) { + if (goodsCoordinator != null) { + goodsCoordinator.markUnavailable(tid, goods.getId()); + } + if (accountTag.isEmpty()) { + log.info("货品 {} 返回「此订单已被抢订」，本批次同租户不再尝试", goods.getId()); + } else { + log.info("{} 货品 {} 返回「此订单已被抢订」，本批次同租户不再尝试", + accountTag, goods.getId()); + } + } else if (goodsCoordinator != null && accountId != null) { + goodsCoordinator.release(tid, goods.getId(), accountId); + } if (buyResult.dailyLimitExceeded()) { stoppedByDailyLimit = true; if (accountTag.isEmpty()) { @@ -615,14 +678,80 @@ public class LbGoodsServiceImpl extends ServiceImpl impl } break; } + if (buyResult.activityNotStarted()) { + stoppedByActivityNotStarted = true; + if (goodsCoordinator != null) { + goodsCoordinator.markActivityNotStarted(); + } + if (accountTag.isEmpty()) { + log.info("活动未开始，停止继续抢购"); + } else { + log.info("{} 活动未开始，停止继续抢购", accountTag); + } + break; + } + if (buyResult.loginRequired()) { + if (tokenRefreshContext != null && !tokenRefreshAttempted) { + if (accountTag.isEmpty()) { + log.info("token 失效或未登录，尝试自动登录刷新 token_front"); + } else { + log.info("{} token 失效或未登录，尝试自动登录刷新 token_front", accountTag); + } + var newTokenOpt = lbBuyAccountTokenService.loginAndPersistTokenFront( + tokenRefreshContext.accountId(), + tid, + tokenRefreshContext.loginAccount(), + tokenRefreshContext.loginPassword()); + if (newTokenOpt.isPresent()) { + tokenRefreshAttempted = true; + tokenRefreshed = true; + apiContext = withToken(apiContext, newTokenOpt.get()); + if (!details.isEmpty()) { + details.remove(details.size() - 1); + } + details.removeIf(d -> !Boolean.TRUE.equals(d.get("success"))); + skippedCount = 0; + failCount = 0; + restartAfterTokenRefresh = true; + if (accountTag.isEmpty()) { + log.info("token 已刷新，重新抢购所有货品"); + } else { + log.info("{} token 已刷新，重新抢购所有货品", accountTag); + } + break; + } + } + if (buyResult.loginRequired()) { + stoppedByLoginRequired = true; + if (accountTag.isEmpty()) { + log.info("token 失效或未登录，停止继续抢购"); + } else { + log.info("{} token 失效或未登录，停止继续抢购", accountTag); + } + break; + } + } } } + } while (restartAfterTokenRefresh); result.put("success", successCount > 0); - if (stoppedByDailyLimit) { + if (stoppedByLoginRequired) { + String loginMsg = tokenRefreshed + ? "token 已自动刷新但仍未登录，已停止抢购" + : HxrAdminBuyService.MSG_LOGIN_REQUIRED; + result.put("message", successCount > 0 + ? loginMsg + "；成功 " + successCount + " 笔，失败 " + failCount + " 笔" + : loginMsg); + } else if (stoppedByDailyLimit) { result.put("message", successCount > 0 ? "当日可抢订单数已满，已停止抢购；成功 " + successCount + " 笔，失败 " + failCount + " 笔" : HxrAdminBuyService.MSG_DAILY_LIMIT_EXCEEDED); + } else if (stoppedByActivityNotStarted) { + result.put("message", successCount > 0 + ? HxrAdminBuyService.MSG_ACTIVITY_NOT_STARTED + + "，已停止抢购；成功 " + successCount + " 笔，失败 " + failCount + " 笔" + : HxrAdminBuyService.MSG_ACTIVITY_NOT_STARTED); } else { result.put("message", successCount > 0 ? "抢购完成，成功 " + successCount + " 笔，失败 " + failCount + " 笔" @@ -630,16 +759,20 @@ public class LbGoodsServiceImpl extends ServiceImpl impl } result.put("successCount", successCount); result.put("failCount", failCount); + result.put("skippedCount", skippedCount); result.put("attemptCount", attemptCount); result.put("maxAttempts", maxAttempts); result.put("stoppedByMax", stoppedByMax); result.put("stoppedByMaxAttempts", stoppedByMaxAttempts); result.put("stoppedByDailyLimit", stoppedByDailyLimit); + result.put("stoppedByLoginRequired", stoppedByLoginRequired); + result.put("stoppedByActivityNotStarted", stoppedByActivityNotStarted); + result.put("tokenRefreshed", tokenRefreshed); result.put("details", details); if (accountTag.isEmpty()) { - log.info("抢购结束，success={}，message={}", successCount > 0, result.get("message")); + log.info("message={},抢购结束，success={}", result.get("message"),successCount > 0); } else { - log.info("{} 抢购结束，success={}，message={}", accountTag, successCount > 0, result.get("message")); + log.info("message={}, {} 抢购结束，success={}", result.get("message"), accountTag, successCount > 0); } return result; } finally { @@ -668,6 +801,17 @@ public class LbGoodsServiceImpl extends ServiceImpl impl return rushBuyAccountLabel.trim(); } + private static HxrGoodsApiContext withToken(HxrGoodsApiContext ctx, String token) { + return new HxrGoodsApiContext( + ctx.goodsApiBaseUrl(), + ctx.buyApiUrl(), + ctx.pageLimit(), + ctx.origin(), + ctx.referer(), + token, + ctx.appStr()); + } + private static void applyDefaults(LbGoods entity) { if (entity.getPrice() == null) { entity.setPrice(BigDecimal.ZERO);