Spring Batch分区场景下请求限流与重试优化方案咨询

我使用Spring Batch通过HTTP API从客户站点获取数据，流程包含两个主要步骤：

• 从API获取文档总数，结合可配置的每页大小计算总页数，通过自定义Paritioner将每页分配给一个分区步骤；
• 分区步骤发送请求获取单页数据（文档列表），处理后写入存储。
  客户站点可能较为“脆弱”，存在限流机制或在大量请求后无响应。

我使用spring-retry对因限流或服务器错误失败的请求进行重试，示例代码如下：

// 分区步骤的ItemReader
@StepScope
public class CustomItemReader extends ItemReader<Object> {
  private List<Object> items;
  @Override
  public Object read() {
    if (Objects.isNull(items)) {
      this.items = ImportService.getPage(pageId);
    }
    if (Objects.nonNull(items) && !items.isEmpty()) {
      return items.remove(0);
    }
    return null;
  }
}

// 配置请求方法的重试逻辑
public class ImportService {
  @Retryable(
      value = RetryableException.class,
      maxAttempts = 3,
      backoff = @Backoff(
          delay = 1000
      )
  )
  public static List<Object> getPage(String pageId) throws RetryableException {
    return ...;
  }
}

重试配置包含Backoff策略，设置了1000ms的递增延迟，以此同时处理重试和限流问题。

• Retryable会持续等待并重执行函数，全程占用线程，数据量较大时实例可能崩溃；
• 每个客户站点有各自的限流规则，使用带Backoff的Retryable并非理想的限流方式，即使为每个客户配置core_pool_size，core_pool_size=1对部分场景也不够用。

1. 有没有合适的方法来限制Spring Batch（尤其是分区场景下）的执行速率？例如配置10秒内发送2次请求，且不能通过在步骤监听器中使用sleep实现。
2. 我曾使用scrapy爬虫，它具备出色的重试和限流功能：RetryMiddleware会将失败页面加入队列并可设置RETRY_LIMIT；AutoThrottle能根据服务器负载自动调整速率。能否在Spring Batch中实现类似功能？还是必须改用scrapy重写项目？

一、Spring Batch分区场景下的速率限制 #

1. 基于Guava RateLimiter的细粒度限流 #

针对每个客户站点单独维护RateLimiter实例，在请求API前先获取许可，相比sleep更灵活，支持动态调整速率，且不会无意义占用线程（可结合tryAcquire()实现非阻塞等待）。

示例改造代码：

// 为每个客户维护独立的RateLimiter
public class RateLimiterManager {
    private static final Map<String, RateLimiter> CLIENT_RATE_LIMITERS = new ConcurrentHashMap<>();

    // 根据客户ID和速率初始化/获取RateLimiter，10秒2次请求对应0.2 permitsPerSecond
    public static RateLimiter getRateLimiter(String clientId, double permitsPerSecond) {
        return CLIENT_RATE_LIMITERS.computeIfAbsent(clientId, k -> RateLimiter.create(permitsPerSecond));
    }
}

// 修改ImportService的getPage方法
public class ImportService {
    @Retryable(
        value = RetryableException.class,
        maxAttempts = 3,
        backoff = @Backoff(delay = 1000)
    )
    public static List<Object> getPage(String pageId, String clientId) throws RetryableException {
        RateLimiter rateLimiter = RateLimiterManager.getRateLimiter(clientId, 0.2);
        rateLimiter.acquire(); // 获取许可后再发起请求
        return ...; // 执行API请求逻辑
    }
}

2. 分区线程池隔离 #

为每个客户站点配置独立的TaskExecutor，结合RateLimiter实现双重控制。在自定义Paritioner中，根据客户ID将任务分配到对应线程池，避免不同客户的请求互相干扰。

二、实现类似Scrapy的重试与自动限流功能 #

1. 类似RetryMiddleware的失败队列重试 #

放弃同步重试逻辑，将失败的页面对象（包含pageId、clientId、当前重试次数）放入延迟队列（可用Redis ZSet实现分布式延迟队列，或本地DelayQueue），再通过独立的Spring Batch作业/定时任务消费队列处理重试，同时设置最大重试次数。

核心思路：

• 改造getPage方法，请求失败时将任务推入延迟队列，抛出非重试异常让当前分区步骤跳过该页面；
• 编写专用Step从延迟队列取任务，检查重试次数是否超限，未超限则重新执行请求；
• 用@Scheduled定时触发重试Step，或通过Spring Batch的JobLauncher自动启动重试任务。

2. 类似AutoThrottle的动态速率调整 #

通过监控请求响应时间、错误率（如429限流码、5xx错误占比），动态调整RateLimiter的速率：

• 用AOP拦截getPage方法，记录每个客户的请求指标；
• 定时统计指标，根据规则调整速率（如错误率超10%则降速20%，响应时间稳定且小于200ms则提速20%）。

示例简化代码：

// 请求监控切面
@Aspect
public class RequestMonitorAspect {
    private Map<String, RequestMetrics> clientMetrics = new ConcurrentHashMap<>();

    @Around("execution(* ImportService.getPage(..))")
    public Object monitorRequest(ProceedingJoinPoint joinPoint) throws Throwable {
        String clientId = (String) joinPoint.getArgs()[1];
        long startTime = System.currentTimeMillis();
        try {
            Object result = joinPoint.proceed();
            clientMetrics.computeIfAbsent(clientId, k -> new RequestMetrics()).recordSuccess(System.currentTimeMillis() - startTime);
            return result;
        } catch (RetryableException e) {
            clientMetrics.computeIfAbsent(clientId, k -> new RequestMetrics()).recordFailure();
            throw e;
        }
    }

    // 每分钟调整一次速率
    @Scheduled(fixedRate = 60000)
    public void adjustRateLimiters() {
        for (Map.Entry<String, RequestMetrics> entry : clientMetrics.entrySet()) {
            String clientId = entry.getKey();
            RequestMetrics metrics = entry.getValue();
            RateLimiter limiter = RateLimiterManager.getRateLimiter(clientId, 0.2);
            double currentRate = limiter.getRate();

            if (metrics.getFailureRate() > 0.1) {
                limiter.setRate(currentRate * 0.8); // 错误率超10%，降速20%
            } else if (metrics.getAvgResponseTime() < 200) {
                limiter.setRate(currentRate * 1.2); // 响应快且稳定，提速20%
            }
            metrics.reset(); // 重置周期指标
        }
    }
}

// 请求指标统计类
class RequestMetrics {
    private int successCount;
    private int failureCount;
    private long totalResponseTime;

    // 省略统计、计算失败率/平均响应时间、重置指标的方法
}

要不要改用Scrapy？ #

如果你的需求核心是纯爬虫场景，Scrapy在重试、限流、爬虫中间件等方面确实更成熟，开发效率更高。但如果项目已基于Spring生态（比如需要和Spring Boot/Spring Cloud集成，或已有其他Spring Batch任务），完全可以在Spring Batch中实现上述功能，不需要重写项目。

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