Java服务实现每秒请求数控制与故障转移的最优方案咨询

一、精准流量控制实现 #

断路器核心作用是故障熔断，无法实现细粒度的QPS精准控制。要达成每秒125条的发送限制，令牌桶算法是最优选择，Guava的RateLimiter轻量易用，可直接满足需求：

• 令牌桶每秒生成125个令牌，每个请求需消耗1个令牌
• 令牌不足时，请求会阻塞等待（也可配置直接拒绝，根据业务需求调整）

以下是修改后的服务实现，整合限流逻辑：

@Slf4j
public class StandardApiCallServiceImpl implements ApiCallService {
    
    private static final int HAPI_HTTP_CODE_NOT_MODIFIED = 1304;
    // 初始化令牌桶，每秒生成125个令牌
    private final RateLimiter rateLimiter = RateLimiter.create(125.0);
    private final ServiceLocator serviceLocator;
    private final EntityCacheService entityCacheService;
    // 存放失败待重试的任务，线程安全队列
    private final Queue<EntityCache> retryQueue = new ConcurrentLinkedQueue<>();

    public StandardApiCallServiceImpl(ServiceLocator serviceLocator, EntityCacheService entityCacheService) {
        this.serviceLocator = serviceLocator;
        this.entityCacheService = entityCacheService;
        // 启动独立重试线程，处理失败任务
        startRetryWorker();
    }

    @Override
    public void makeApiCall(List<EntityCache> entityCaches) {
        for (EntityCache entity : entityCaches) {
            // 阻塞等待获取令牌，确保QPS不超过限制
            rateLimiter.acquire();
            manageEntityCache(entity);
        }
    }

    protected void manageEntityCache(final EntityCache entityCache) {
        // 整合Resilience4j断路器，实现故障熔断
        CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("third-party-api-circuit");
        Try.ofSupplier(() -> circuitBreaker.executeSupplier(() -> send(entityCache)))
                .onSuccess(response -> {
                    updateExecutionInformation(
                            entityCache,
                            calculateStatus(entityCache, response),
                            response
                    );
                })
                .onFailure(HapiCacheEntityException.class, e -> {
                    updateExecutionInformation(entityCache, StatusType.CACHED, VioohResponse.builder()
                            .isSuccessful(true)
                            .httpCode(HAPI_HTTP_CODE_NOT_MODIFIED)
                            .build()
                    );
                    log.info("entity was cached and was not resent by hapi: {}", entityCache);
                })
                .onFailure(e -> {
                    log.error("error occurred when trying to send this entity cache: {}, error: {}", entityCache, e.getMessage());
                    // 失败任务加入重试队列
                    retryQueue.offer(entityCache);
                });
    }

    protected Response send(final EntityCache entityCache) {
        final EntityType entityType = EntityType.valueOf(entityCache.getEntityType().getCode());
        // 整合Resilience4j重试，处理瞬时故障
        Retry retry = Retry.ofDefaults("third-party-api-retry");
        return retry.executeSupplier(() -> 
            serviceLocator.find(entityType.getFamilyType())
                .send(entityType, entityCache)
        );
    }

    // 重试工作线程，定期重试失败任务
    private void startRetryWorker() {
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            while (!retryQueue.isEmpty()) {
                EntityCache entity = retryQueue.poll();
                if (entity != null) {
                    rateLimiter.acquire(); // 重试请求同样受限流控制
                    try {
                        manageEntityCache(entity);
                    } catch (Exception e) {
                        // 多次重试失败后，持久化到数据库/消息队列，避免数据丢失
                        log.error("retry failed for entity: {}, error: {}", entity, e.getMessage());
                        // 示例：调用缓存服务持久化失败任务
                        // entityCacheService.saveFailedEntity(entity);
                    }
                }
            }
        }, 1, 5, TimeUnit.SECONDS); // 每5秒执行一次重试逻辑
    }

    // 原有calculateStatus、updateExecutionInformation方法保持不变
}

二、故障转移机制优化 #

1. 断路器（Resilience4j）：
   • 配置错误率阈值（如50%）或连续失败次数，触发断路器打开，停止向故障服务发送请求，避免雪崩
   • 半开状态下尝试发送少量请求验证服务状态，恢复正常后自动关闭断路器
2. 重试机制：
   • 针对网络抖动、瞬时超时等临时故障，配置重试次数、间隔，自动重试请求
3. 失败任务持久化：
   • 多次重试仍失败的任务，存入数据库或消息队列，后续通过定时任务或人工介入处理，避免数据丢失
4. 监控告警：
   • 对接监控系统，当断路器打开、失败任务堆积时触发告警，及时排查问题

三、关键注意事项 #

• 限流逻辑需覆盖所有请求路径（包括重试请求），确保整体QPS符合要求
• 重试线程需独立于主业务线程，避免阻塞正常业务流程
• 根据第三方服务的稳定性、响应时间，调整断路器阈值、重试次数等参数

内容的提问来源于stack exchange，提问作者L.r. Sharma