二手房网站排行做引流推广的平台
目录
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 "{}";}}}