目錄
1、并發限速
2、串行限速??
?
需求:批量調用第三方ERP接口,對方接口限流時,減緩調用速率。
1、并發限速
@Slf4j
@RestController
public class ApiCallTask {//第三方接口@Resourceprivate ErpService erpService;//異步線程池@Resourceprivate ThreadPoolTaskExecutor taskExecutor;//定時調度器@Resourceprivate ThreadPoolTaskScheduler taskScheduler;private static final BlockingQueue<Seller> sellerQueue = new LinkedBlockingQueue<>(1000);private static final RateLimiter rateLimiter = RateLimiter.create(200.0 / 60.0);private static final ObjectMapper objectMapper = new ObjectMapper();private static final int MAX_RETRY_COUNT = 5;private static final int BATCH_SIZE = 10;@Scheduled(cron = "0 0 2 * * ?")@RequestMapping(value = "/jobAfterSalesSync")public ResponseEntity<String> jobAfterSalesSync() {log.info("開始同步商家售后數據...");Map<String, String> queryMap = Maps.newHashMap();queryMap.put("status", "2");List<商家seller> sellerList = erpService.getSellerList(queryMap);List<商家seller> sellerList = sellerList != null ? sellerList : new ArrayList<>();log.info("共 {} 個商家待處理", sellerList.size());for (Seller seller : sellerList) {if (!sellerQueue.offer(seller)) {log.warn("隊列已滿,商家 {} 未加入隊列", seller.getSellerName());} else {log.debug("商家 {} 已加入隊列", seller.getSellerName());}}processBatch(); // 啟動分批處理log.info("任務已提交,線程池活躍線程數: {}", taskExecutor.getActiveCount());return ResponseEntity.ok("任務已觸發");}/*** 異步處理任務* taskScheduler與rateLimiter的分工* processBatch 中每批完成后等待 3 秒再調度下一批,這是批次之間的宏觀控制。Instant.now().plusSeconds(3)* rateLimiter.acquire():在每批內部的 10 個任務中,控制每個 API 調用的微觀速率。* 隊列作用* processBatch 在每次批次完成后檢查 sellerQueue.isEmpty()。如果隊列非空,通過 taskScheduler.schedule 調度下一次 processBatch,形成遞歸調用。保證所有seller都被處理* 限流(801)時,handleRetry 確保 Seller 被重新加入 sellerQueue。即使隊列滿,也通過延遲重試保證任務不丟失。* CompletableFuture作用* CompletableFuture 是對傳統 Future 的增強,支持鏈式調用、異常處理和任務組合,用于異步執行 callErpApi,實現每批 10 個 Seller 的并發處理。* 將 callErpApi 的執行從主線程中分離出來,提交給線程池(如 taskExecutor)異步運行。* submit 方法返回一個 Future 對象(這里未使用返回值),表示任務已交給線程池處理。* CompletableFuture.runAsync 創建異步任務,執行 callErpApi。在批處理中,每個 Seller 的 API 調用是獨立的異步任務。* futures 收集所有任務的 CompletableFuture 實例。* 使用 CompletableFuture.allOf 等待一批任務全部完成,然后觸發后續操作(如調度下一批)。* 通過 .exceptionally 或 .whenComplete 處理異步任務中的異常,確保任務鏈不會因錯誤中斷。* taskExecutor 的整體作用* 異步執行:* 將 processBatch 和 callErpApi 從主線程(定時任務或 HTTP 請求線程)中分離出來,避免阻塞主線程。* 例如,HTTP 請求可以快速響應,而實際處理在后臺進行。* 并發處理:* 在 processBatch 中,10 個 callErpApi 任務可以并行執行(取決于線程池大小),提高處理效率。* 例如,如果線程池核心線程數為 10,則每批 10 個任務可以同時運行。* 與 taskScheduler 的分工:* taskExecutor:負責執行具體的任務(processBatch 和 callErpApi)。* taskScheduler:負責調度任務的執行時間(例如批次間隔 3 秒或限流重試延遲)。* 與代碼目標的關系* 分批執行:taskExecutor 使每批 10 個任務并發運行。* 持續執行:與 taskScheduler 配合,確保隊列非空時任務持續調度。* 限流控制:rateLimiter 限制速率,taskExecutor 提供并發支持,二者結合實現高效且受控的處理。*/private void processBatch() {if (sellerQueue.isEmpty()) {log.info("隊列處理完成,剩余大小: {}", sellerQueue.size());return;}List<Seller> batch = new ArrayList<>();int drained = sellerQueue.drainTo(batch, BATCH_SIZE); // 取出最多 10 個,出隊log.info("從隊列中取出 {} 個元素,開始處理", drained);List<CompletableFuture<Void>> futures = new ArrayList<>();/**** 對于每批的 10 個 Seller,使用 CompletableFuture.runAsync 將 callErpApi(seller) 提交到 taskExecutor 執行。* runAsync 的第二個參數指定了執行器(taskExecutor),確保這些任務在 taskExecutor 的線程池中并行運行。*/for (Seller seller : batch) {CompletableFuture<Void> future = CompletableFuture.runAsync(() -> callErpApi(seller), taskExecutor);futures.add(future);}// 等待當前批次所有任務完成。在批次完成后執行回調,檢查隊列并調度下一次 processBatch(延遲 3 秒)。CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).whenComplete((result, throwable) -> {if (throwable != null) {log.error("批處理異常: {}", throwable.getMessage());}log.info("完成一批處理,剩余隊列大小: {}", sellerQueue.size());if (!sellerQueue.isEmpty()) {// 只要 sellerQueue 中還有元素,processBatch 會在每次批次完成后通過 taskScheduler.schedule 重新調用自己。//限流(801 錯誤)時,handleRetry 會將 Seller 重新加入 sellerQueue,保持隊列非空。//每次批次完成后,只要隊列非空,就延遲 3 秒調度下一批,直到隊列為空。taskScheduler.schedule(this::processBatch, Instant.now().plusSeconds(3));} else {log.info("隊列已空,任務結束");}});}//調用ERP接口private void callErpApi(Seller seller) {rateLimiter.acquire();try {String response = request(seller);if (StringUtils.isBlank(response)) {log.info("商家 {} 處理成功(隊列剩余: {})", seller.getSellerName(), sellerQueue.size());} else {JsonNode jsonResponse = objectMapper.readTree(response);if (jsonResponse.has("code") && jsonResponse.get("code").asInt() == 801) {int waitTime = extractWaitTime(jsonResponse.get("message").asText());log.warn("限流,商家 {} 暫停 {} 秒后重試", seller.getSellerName(), waitTime);//限流時,sellerQueue.offer(seller) 嘗試入隊。如果隊列滿,延遲 waitTime 秒后重試。handleRetry(seller, waitTime);} else {log.warn("其他錯誤,商家: {}, 接口返回: {},視為成功", seller.getSellerName(), response);}}} catch (Exception e) {log.error("API 調用異常,商家: {},視為成功", seller.getSellerName(), e);}}//重試處理private void handleRetry(Seller seller, int waitTime) {if (seller.getRetryCount() < MAX_RETRY_COUNT) {seller.incrementRetry();boolean requeued = sellerQueue.offer(seller);//入隊if (requeued) {log.info("商家 {} 重試次數: {},已重新入隊,等待下次批處理", seller.getSellerName(), seller.getRetryCount());} else {log.warn("隊列已滿,商家 {} 延遲 {} 秒后重試", seller.getSellerName(), waitTime);taskScheduler.schedule(() -> handleRetry(seller, waitTime), Instant.now().plusSeconds(waitTime));}} else {log.error("商家 {} 達到最大重試次數 {},丟棄", seller.getSellerName(), MAX_RETRY_COUNT);}}//獲取限速接口中等待時間private static int extractWaitTime(String message) {Pattern pattern = Pattern.compile("(\\d+)\\s*秒");Matcher matcher = pattern.matcher(message);return matcher.find() ? Integer.parseInt(matcher.group(1)) : 30;}//請求接口public String request(Seller seller) {Map<String, Object> params = Maps.newHashMap();String[] range = DateUtil.getDateRange(14);params.put("limit", 200);params.put("page", 1);params.put("start_time", range[0]);params.put("end_time", range[1]);params.put("shop_nick", seller.getSellerName());try {String result = erpService.afterSalesData(params);return result != null ? result : "";} catch (Exception e) {log.error("請求 API 失敗,商家: {}", seller.getSellerName(), e);return "{}";}}}
2、串行限速??
以上代碼仍然有限速問題,調用接口限速頻率太高,改造并優化。
@Slf4j
@RestController
public class ErpApiCallTask {@Resourceprivate ErpService erpService;@Resourceprivate ThreadPoolTaskExecutor taskExecutor;@Resourceprivate ThreadPoolTaskScheduler taskScheduler;private static final BlockingQueue<Seller> sellerQueue = new LinkedBlockingQueue<>(1000);private static final RateLimiter rateLimiter = RateLimiter.create(1.0 / 5.0); // 每 5 秒 1 次private static final ObjectMapper objectMapper = new ObjectMapper();private static final int MAX_RETRY_COUNT = 5;private static final long WAIT_INTERVAL = 30000; // 等待 30 秒檢查新入隊元素@Scheduled(cron = "0 0 2 * * ?")@RequestMapping(value = "/jobAfterSalesSync")public ResponseEntity<String> jobAfterSalesSync() {log.info("開始從ERP系統同步商家售后數據...");Map<String, String> queryMap = Maps.newHashMap();queryMap.put("status", "2");List<商家seller> sellerList = erpService.getSellerList(queryMap);log.info("共 {} 個商家待處理", sellerList.size());for (Seller seller : sellerList) {if (!sellerQueue.offer(seller)) {log.warn("隊列已滿,商家 {} 未加入隊列,當前隊列大小: {}", seller.getSellerName(), sellerQueue.size());} else {log.debug("商家 {} 已加入隊列,當前隊列大小: {}", seller.getSellerName(), sellerQueue.size());}}log.info("隊列初始化完成,當前隊列大小: {}", sellerQueue.size());// 異步啟動處理taskExecutor.submit(this::processQueue);log.info("任務已提交,線程池活躍線程數: {}", taskExecutor.getActiveCount());return ResponseEntity.ok("任務已觸發");}/*** 串行處理隊列,使用 RateLimiter 控制每 5 秒 1 次調用,隊列為空時等待新入隊元素* 移除批處理和并發:原代碼按批次處理(每次 10 個),并通過 CompletableFuture 并發執行。現在改為 processQueue,串行處理隊列中的每個 Seller。* 移除 CompletableFuture:不需要并發,直接在單線程中順序調用 callErpApi。* 串行執行:使用 orderQueue.poll() 逐個取出 Seller,每次處理一個后等待 5 秒。* 保留 taskExecutor.submit:異步啟動處理,避免阻塞主線程(定時任務或 HTTP 請求)。處理邏輯在后臺線程中串行執行。* 使用 RateLimiter 控制速率,在每次調用 callErpApi 前獲取令牌,確保5秒最多 1 次調用。* 相比 Thread.sleep(1000),RateLimiter 更靈活,能動態調整速率并處理突發請求。* 限流處理,限流(801)時,延遲 waitTime 秒后重新入隊。隊列滿時遞歸重試,確保任務不丟失。* 新增等待機制:當 orderQueue.isEmpty() 時,不直接退出,而是等待 WAIT_INTERVAL(秒),然后再次檢查隊列。如果等待后隊列仍為空,設置 hasMoreTasks = false,結束循環;否則繼續處理。* 新增標志變量 hasMoreTasks:用布爾變量控制外層循環,避免無限等待。*/private void processQueue() {boolean hasMoreTasks = true;while (hasMoreTasks) {if (!sellerQueue.isEmpty()) {log.info("開始處理隊列,當前隊列大小: {}", sellerQueue.size());Seller seller = sellerQueue.poll(); // 取出隊列頭部元素,if (seller != null) {log.info("從隊列中取出商家: {},剩余隊列大小: {}", seller.getSellerName(), sellerQueue.size());rateLimiter.acquire(); // 獲取令牌,控制速率callErpApi(seller);}} else {log.info("隊列當前為空,等待 {} 毫秒檢查新入隊元素", WAIT_INTERVAL);try {Thread.sleep(WAIT_INTERVAL); // 等待一段時間,檢查是否有新元素} catch (InterruptedException e) {log.error("等待被中斷", e);Thread.currentThread().interrupt();}if (sellerQueue.isEmpty()) {log.info("等待后隊列仍為空,任務結束");hasMoreTasks = false; // 隊列仍為空,結束循環} else {log.info("檢測到新入隊元素,繼續處理,當前隊列大小: {}", sellerQueue.size());}}}log.info("隊列處理完成,剩余大小: {}", sellerQueue.size());}private void callErpApi(Seller seller) {try {String response = request(seller);if (StringUtils.isBlank(response)) {log.info("商家 {} 處理成功,當前隊列大小: {}", seller.getSellerName(), sellerQueue.size());} else {JsonNode jsonResponse = objectMapper.readTree(response);if (jsonResponse.has("code") && jsonResponse.get("code").asInt() == 801) {int waitTime = extractWaitTime(jsonResponse.get("message").asText());log.warn("限流,商家 {} 暫停 {} 秒后重試,當前隊列大小: {}",seller.getSellerName(), waitTime, sellerQueue.size());handleRetry(seller, waitTime);} else {log.warn("其他錯誤,商家: {},接口返回: {},視為成功,當前隊列大小: {}",seller.getSellerName(), response, sellerQueue.size());}}} catch (Exception e) {log.error("API 調用異常,商家: {},視為成功,當前隊列大小: {}",seller.getSellerName(), sellerQueue.size(), e);}}private void handleRetry(Seller seller, int waitTime) {if (seller.getRetryCount() < MAX_RETRY_COUNT) {seller.incrementRetry();// 延遲 waitTime 秒后重新入隊taskScheduler.schedule(() -> {boolean requeued = sellerQueue.offer(seller);if (requeued) {log.info("商家 {} 重試次數: {},已重新入隊,當前隊列大小: {}",seller.getSellerName(), seller.getRetryCount(), sellerQueue.size());} else {log.warn("隊列已滿,商家 {} 延遲 {} 秒后重試,當前隊列大小: {}",seller.getSellerName(), waitTime, sellerQueue.size());handleRetry(seller, waitTime); // 遞歸重試}}, Instant.now().plusSeconds(waitTime));} else {log.error("商家 {} 達到最大重試次數 {},丟棄,當前隊列大小: {}",seller.getSellerName(), MAX_RETRY_COUNT, sellerQueue.size());}}private static int extractWaitTime(String message) {Pattern pattern = Pattern.compile("(\\d+)\\s*秒");Matcher matcher = pattern.matcher(message);return matcher.find() ? Integer.parseInt(matcher.group(1)) : 30;}public String request(Seller seller) {Map<String, Object> params = Maps.newHashMap();String[] range = DateUtil.getDateRange(14);params.put("limit", 200);params.put("page", 1);params.put("start_time", range[0]);params.put("end_time", range[1]);params.put("shop_nick", seller.getSellerName());try {String result = erpService.afterSalesSyncHandler(params);return result != null ? result : "";} catch (Exception e) {log.error("請求 API 失敗,商家: {}", seller.getSellerName(), e);return "{}";}}}