Spring Boot虚拟线程下异步处理速率控制方案咨询

针对Spring Boot 3.2.3 + Java 21虚拟线程环境下，因无限制异步并发调用第三方API触发429 Too Many Requests的问题，以下是几种实用优化方案：

1. 令牌桶限流（精准控制请求速率） #

使用Guava的RateLimiter实现令牌桶算法，严格控制每秒发送给第三方API的请求数，从根源避免触发限流。

实现步骤： #

• 引入Guava依赖（Maven）：

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>32.1.3-jre</version>
</dependency>

• 创建RateLimiter Bean：

@Bean
public RateLimiter thirdPartyApiRateLimiter() {
    // 根据第三方API的限流规则调整速率，示例为每秒100次请求
    return RateLimiter.create(100.0);
}

• 在异步调用方法中加入令牌获取逻辑：

@Service
public class ThirdPartyApiService {
    private final RateLimiter rateLimiter;
    private final RestTemplate restTemplate;

    public ThirdPartyApiService(RateLimiter thirdPartyApiRateLimiter, RestTemplate restTemplate) {
        this.rateLimiter = thirdPartyApiRateLimiter;
        this.restTemplate = restTemplate;
    }

    @Async
    public CompletableFuture<String> callApi(String param) {
        // 阻塞等待获取令牌，也可使用tryAcquire()实现非阻塞逻辑
        rateLimiter.acquire();
        try {
            String response = restTemplate.getForObject(
                "https://third-party-api.com/api?param={param}", 
                String.class, 
                param
            );
            return CompletableFuture.completedFuture(response);
        } catch (Exception e) {
            return CompletableFuture.failedFuture(e);
        }
    }
}

2. 结合重试机制处理429错误 #

即使做了限流，仍可能因第三方API的动态限流策略触发429，此时可通过Spring Retry实现自动重试，提升请求成功率。

实现步骤： #

• 引入Spring Retry依赖（Maven）：

<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>

• 启用重试功能：

@EnableRetry
@SpringBootApplication
public class YourApplication {
    public static void main(String[] args) {
        SpringApplication.run(YourApplication.class, args);
    }
}

• 在异步方法上添加重试注解：

@Async
@Retryable(
    value = {HttpClientErrorException.TooManyRequests.class},
    backoff = @Backoff(delay = 1000, multiplier = 2, maxDelay = 5000) // 指数退避重试策略
)
public CompletableFuture<String> callApi(String param) {
    rateLimiter.acquire();
    try {
        String response = restTemplate.getForObject(
            "https://third-party-api.com/api?param={param}", 
            String.class, 
            param
        );
        return CompletableFuture.completedFuture(response);
    } catch (HttpClientErrorException.TooManyRequests e) {
        // 抛出异常触发重试逻辑
        throw e;
    } catch (Exception e) {
        return CompletableFuture.failedFuture(e);
    }
}

3. 限制虚拟线程并发数 #

虽然虚拟线程轻量，但无限制创建仍会导致第三方API请求过载，可通过Semaphore包装VirtualThreadTaskExecutor，限制最大并发数。

自定义TaskExecutor示例： #

@Bean
public AsyncTaskExecutor applicationTaskExecutor() {
    VirtualThreadTaskExecutor delegate = new VirtualThreadTaskExecutor("async-worker");
    int maxConcurrency = 50; // 根据第三方API并发限制调整
    Semaphore semaphore = new Semaphore(maxConcurrency);

    return new AsyncTaskExecutor() {
        @Override
        public void execute(Runnable task) {
            try {
                semaphore.acquire();
                delegate.execute(() -> {
                    try {
                        task.run();
                    } finally {
                        semaphore.release();
                    }
                });
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IllegalStateException("Failed to acquire semaphore for async task", e);
            }
        }

        @Override
        public <T> Future<T> submit(Callable<T> task) {
            try {
                semaphore.acquire();
                return delegate.submit(() -> {
                    try {
                        return task.call();
                    } finally {
                        semaphore.release();
                    }
                });
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IllegalStateException("Failed to acquire semaphore for async task", e);
            }
        }

        @Override
        public Future<?> submit(Runnable task) {
            return submit(Executors.callable(task));
        }

        @Override
        public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
            try {
                semaphore.acquire();
                return delegate.submitListenable(() -> {
                    try {
                        return task.call();
                    } finally {
                        semaphore.release();
                    }
                });
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IllegalStateException("Failed to acquire semaphore for async task", e);
            }
        }

        @Override
        public ListenableFuture<?> submitListenable(Runnable task) {
            return submitListenable(Executors.callable(task));
        }
    };
}

方案优先级推荐 #

优先选择令牌桶限流 + 重试机制的组合：令牌桶从源头控制请求速率，重试机制处理突发的429错误，两者结合能有效降低第三方API限流触发概率，同时提升请求成功率。

内容的提问来源于stack exchange，提问作者SHM