admin/smartDriveEE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构抢单代码

Browse Source
This commit is contained in:
cst61
2026-06-05 10:12:29 +08:00
parent 1445c01c78
commit 78c1fd60a7
12 changed files with 552 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/java/com/rj/controller/LbBuyAccountController.java
View File

@@ -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<Map<String, Object>> rushBuy(

2
src/main/java/com/rj/controller/LbGoodsController.java
View File

@@ -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")

10
src/main/java/com/rj/dto/LbBuyAccountRushBuyContext.java Normal file
View File

@@ -0,0 +1,10 @@
package com.rj.dto;
/**
* 抢单时 token 失效后自动登录刷新 {@code lb_buy_account.token_front} 所需账号信息。
*/
public record LbBuyAccountRushBuyContext(
String accountId,
String loginAccount,
String loginPassword) {
}

74
src/main/java/com/rj/dto/LbRushBuyGoodsCoordinator.java Normal file
View File

@@ -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<String, String> 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());
}
}

14
src/main/java/com/rj/mapper/LbBuyAccountMapper.java
View File

@@ -43,4 +43,18 @@ public interface LbBuyAccountMapper extends BaseMapper<LbBuyAccount> {
@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);
}

32
src/main/java/com/rj/service/HxrAdminBuyService.java
View File

@@ -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() {

30
src/main/java/com/rj/service/HxrAdminUserLoginService.java
View File

@@ -120,6 +120,36 @@ public class HxrAdminUserLoginService {
return new LoginApiResult(ok, code, msg, parsed, httpStatus);
}
@SuppressWarnings("unchecked")
public static Map<String, Object> 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<String, Object>) userinfoObj;
}
/** 从登录 API 解析结果中提取 {@code data.userinfo.token}。 */
public static String extractToken(Object parsed) {
Map<String, Object> 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,

33
src/main/java/com/rj/service/ILbGoodsService.java
View File

@@ -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<LbGoods> {
Map<String, Object> 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<String, Object> rushBuy(String tenantId, String token, Integer maxBuyCount) {
return rushBuy(tenantId, token, maxBuyCount, null);
return rushBuy(tenantId, token, maxBuyCount, null, null, null, null);
}
Map<String, Object> rushBuy(String tenantId, String token, Integer maxBuyCount, String rushBuyAccountLabel);
default Map<String, Object> rushBuy(String tenantId, String token, Integer maxBuyCount, String rushBuyAccountLabel) {
return rushBuy(tenantId, token, maxBuyCount, rushBuyAccountLabel, null, null, null);
}
default Map<String, Object> rushBuy(
String tenantId,
String token,
Integer maxBuyCount,
String rushBuyAccountLabel,
LbBuyAccountRushBuyContext tokenRefreshContext) {
return rushBuy(tenantId, token, maxBuyCount, rushBuyAccountLabel, tokenRefreshContext, null, null);
}
Map<String, Object> rushBuy(
String tenantId,
String token,
Integer maxBuyCount,
String rushBuyAccountLabel,
LbBuyAccountRushBuyContext tokenRefreshContext,
LbRushBuyGoodsCoordinator goodsCoordinator,
String accountId);
}

97
src/main/java/com/rj/service/LbBuyAccountTokenService.java Normal file
View File

@@ -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<String> 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<HxrUserLoginApiContext> 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) + "...";
}
}

181
src/main/java/com/rj/service/impl/LbBuyAccountServiceImpl.java
View File

@@ -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<Map<String, Object>> accountResults;
int successAccounts;
int failAccounts;
try {
CountDownLatch startLatch = new CountDownLatch(1);
CountDownLatch readyLatch = new CountDownLatch(accountCount);
CountDownLatch doneLatch = new CountDownLatch(accountCount);
@SuppressWarnings("unchecked")
Map<String, Object>[] resultArray = new Map[accountCount];
for (int i = 0; i < accountCount; i++) {
final int index = i;
String id = normalizedIds.get(i);
List<CompletableFuture<Map<String, Object>>> 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<Map<String, Object>> accountResults = new ArrayList<>(accountCount);
int successAccounts = 0;
int failAccounts = 0;
for (Map<String, Object> item : resultArray) {
accountResults = new ArrayList<>(normalizedIds.size());
successAccounts = 0;
failAccounts = 0;
for (CompletableFuture<Map<String, Object>> future : futures) {
Map<String, Object> item = future.join();
accountResults.add(item);
if (Boolean.TRUE.equals(item.get("success"))) {
successAccounts++;
@@ -300,27 +272,29 @@ public class LbBuyAccountServiceImpl
failAccounts++;
}
}
} finally {
executor.shutdown();
try {
if (!executor.awaitTermination(2, TimeUnit.HOURS)) {
log.warn("批量抢单线程池未在 2 小时内结束，执行 shutdownNow");
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
throw new RuntimeException("批量抢单等待线程结束被中断", e);
}
}
result.put("success", successAccounts > 0);
result.put("message", successAccounts > 0
? "批量抢购完成，成功账号 " + successAccounts + " 个，失败 " + failAccounts + " 个"
: "批量抢购未成功");
result.put("totalAccounts", accountCount);
result.put("totalAccounts", normalizedIds.size());
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();
}
}
} catch (Exception e) {
log.error("抢单账号批量抢购异常", e);
result.put("success", false);
@@ -329,6 +303,20 @@ public class LbBuyAccountServiceImpl
}
}
private Map<String, Object> 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<String, Object> item = buildRushBuyTaskErrorItem(id, account, e.getMessage());
persistRushBuyOutcomeFromItem(account, item);
return item;
}
}
private static Map<String, Object> buildRushBuyTaskErrorItem(String accountId, LbBuyAccount account, String reason) {
String accountLabel = formatRushBuyAccountLabel(accountId, account);
Map<String, Object> item = new LinkedHashMap<>();
@@ -344,9 +332,10 @@ public class LbBuyAccountServiceImpl
}
/**
* 单账号抢单逻辑，供线程池并行调用。
* 单账号抢单逻辑；批量入口为每账号一线程并发调用，共享 {@link LbRushBuyGoodsCoordinator}。
*/
private Map<String, Object> processSingleAccountRushBuy(String id, LbBuyAccount account) {
private Map<String, Object> processSingleAccountRushBuy(
String id, LbBuyAccount account, LbRushBuyGoodsCoordinator goodsCoordinator) {
String accountLabel = formatRushBuyAccountLabel(id, account);
log.info("抢单任务开始，{}，thread={}", accountLabel, Thread.currentThread().getName());
Map<String, Object> 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<String, Object> 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);
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<String, Object> 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<String, Object> rushBuyResult, Object summaryMsg) {
Boolean overallSuccess = rushBuyResult.get("success") instanceof Boolean

18
src/main/java/com/rj/service/impl/LbFanManagementServiceImpl.java
View File

@@ -305,7 +305,7 @@ public class LbFanManagementServiceImpl
item.put("parsed", loginResult.parsed());
if (loginResult.success()) {
successCount++;
Map<String, Object> userinfo = extractUserinfo(loginResult.parsed());
Map<String, Object> userinfo = HxrAdminUserLoginService.extractUserinfo(loginResult.parsed());
if (userinfo != null) {
Map<String, Object> saveResult =
saveFanFromLoginUserinfo(userinfo, tenantId);
@@ -438,22 +438,6 @@ public class LbFanManagementServiceImpl
}
}
@SuppressWarnings("unchecked")
private static Map<String, Object> 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<String, Object>) userinfoObj;
}
private Map<String, Object> saveFanFromLoginUserinfo(Map<String, Object> userinfo, String tenantId) {
Map<String, Object> saveResult = new LinkedHashMap<>();
Long fanId = parseLong(userinfo.get("id"));

156
src/main/java/com/rj/service/impl/LbGoodsServiceImpl.java
View File

@@ -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<LbGoodsMapper, LbGoods> 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<LbGoodsMapper, LbGoods> impl
private final ILbThirdIntegrationConfigService lbThirdIntegrationConfigService;
private final LbBuyAccountTokenService lbBuyAccountTokenService;
@Override
public Map<String, Object> add(LbGoods entity) {
Map<String, Object> result = new HashMap<>();
@@ -513,7 +519,14 @@ public class LbGoodsServiceImpl extends ServiceImpl<LbGoodsMapper, LbGoods> impl
}
@Override
public Map<String, Object> rushBuy(String tenantId, String token, Integer maxBuyCount, String rushBuyAccountLabel) {
public Map<String, Object> rushBuy(
String tenantId,
String token,
Integer maxBuyCount,
String rushBuyAccountLabel,
LbBuyAccountRushBuyContext tokenRefreshContext,
LbRushBuyGoodsCoordinator goodsCoordinator,
String accountId) {
Map<String, Object> result = new HashMap<>();
String accountTag = formatRushBuyAccountLogTag(rushBuyAccountLabel);
try {
@@ -550,28 +563,41 @@ public class LbGoodsServiceImpl extends ServiceImpl<LbGoodsMapper, LbGoods> 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<LbGoods> 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<Map<String, Object>> 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<LbGoodsMapper, LbGoods> impl
continue;
}
if (goodsCoordinator != null && accountId != null
&& !goodsCoordinator.tryAcquire(tid, goods.getId(), accountId)) {
skippedCount++;
Map<String, Object> 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<String, Object> item = new LinkedHashMap<>();
item.put("id", goods.getId());
@@ -604,8 +653,22 @@ public class LbGoodsServiceImpl extends ServiceImpl<LbGoodsMapper, LbGoods> 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<LbGoodsMapper, LbGoods> 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<LbGoodsMapper, LbGoods> 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<LbGoodsMapper, LbGoods> 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);