分区处理“长短腿”：Spring Batch 数据倾斜让你的批处理永远在等最慢的那个分区

你用 Spring Boot + Spring Batch 搭建了批处理任务，数据量上亿，信心满满地配了分区并行，心想几十个 Worker 一起跑，速度肯定起飞。结果执行下来，大部分 Worker 几分钟就跑完，却总有一个“拖后腿”的分区还在苦哈哈跑几小时，整个 Job 的总耗时被这个最慢分区死死钉住。你一看数据分布，那个分区的数据量是别的分区的几十倍，甚至包含了全公司最大的客户数据——这就是数据倾斜，分布式计算的死敌，同样也是 Spring Batch 分区处理中最隐蔽的性能杀手。

本文将深挖 Spring Batch 分区处理中数据倾斜的成因与破坏力，从自定义分区键、动态分区、远程分区负载均衡，到结合数据库分片与 Redis 缓存的根治方案，给你一套让分区负载均衡、不再“苦等最慢一个”的实战工具箱。

---

一、血泪现场：数据倾斜的三幕“苦情戏”

1.1 按 ID 取模分区，大客户数据全部落入一个分区

你有一张 orders 表，包含普通用户和少数大客户（如平台自营店铺）。你用 orderId % 10 作为分区键，10 个分区并行处理。结果自营店铺的订单 ID 哈希到同一个分区，该分区数据量是其他分区的 20 倍，其他分区早早空闲，整体 Job 时间被这个“天选分区”拖垮。

1.2 按日期分区，某一天的数据量特别大

促销活动当天订单量是平时的 30 倍，你按 create_date 字段范围分区，每个分区处理一天。活动日分区单独跑了 8 小时，其他分区总计不到 2 小时。Job 耗时取决于最慢的那一天。

1.3 按用户 ID 分片，少量用户拥有海量操作记录

你按 userId.hashCode() % 64 分区，批处理用户行为日志。结果，少量爬虫用户或系统账号产生了上千万条记录，集中到几个分区，导致这些分区所在的 Worker CPU 和内存爆满，甚至 OOM。

这些故事都在传达同一个信息：不合理的分区策略无法均匀分割数据，而 Spring Batch 的 StepExecutionSplitter 默认只是机械地分配分区，不关心数据分布。一旦倾斜，并行化带来的加速就大幅退化，甚至不如单机顺序处理。

---

二、根因剖析：Spring Batch 分区模型与倾斜的本质

Spring Batch 的分区分为两种：

• 本地分区：TaskExecutorPartitionHandler，在主节点启动多个线程执行从 step。
• 远程分区：MessageChannelPartitionHandler 或 DeployerPartitionHandler，将分区任务分发到其他 JVM 或容器。

无论是哪种，核心都是：

1. Partitioner 接口的 partition(int gridSize) 方法生成若干个 StepExecution，每个分配一个分区上下文（ExecutionContext）。
2. 从 step 读取该上下文中的分区范围（例如 minId 和 maxId），各自独立处理。
3. 所有从 step 都完成后，主 step 才算完成。

数据倾斜的实质：Partitioner 生成的范围大小严重不均，导致不同分区的工作量差异巨大。默认的 MultiResourcePartitioner 或 ColumnRangePartitioner 按照资源数量或值的范围等分，但不考虑数据密度的差异。若数据分布不均（如 ID 不连续、热点数据），分区大小就会失衡。

解决倾斜，必须让 Partitioner 能够根据真实数据分布，自适应地划分出大小相近的子集。

---

三、解决方案一：自定义 Partitioner，使用基于采样的动态分区

目标：不再机械地按值范围均匀切分，而是先对数据分布进行采样，再生成分区边界，使每个分区的数据量大致相等。

3.1 实现思路

• 在 Partitioner 中连接数据库，执行分组查询，获取分布直方图。
• 根据 gridSize 和目标数据量，计算每个分区的边界（如主键范围）。
• 返回的 Map<String, ExecutionContext> 只包含大致等量的数据段。

3.2 代码示例：基于用户 ID 分区的动态范围划分

假设 orders 表有 id（bigint）和 user_id，我们希望按 id 范围分区，但很多 ID 段没有数据（稀疏），直接等距分会造成空分区或倾斜。

@Component
public class DynamicRangePartitioner implements Partitioner {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        // 1. 查询总记录数
        Long total = jdbcTemplate.queryForObject("SELECT COUNT(*) FROM orders", Long.class);
        long perPartition = total / gridSize; // 每个分区期望的记录数

        // 2. 采样：查询有序的id，每隔 perPartition 取一个边界 id
        List<Long> boundaries = jdbcTemplate.queryForList(
            "SELECT id FROM (SELECT id, ROW_NUMBER() OVER (ORDER BY id) AS rn FROM orders) tmp " +
            "WHERE MOD(rn, ?) = 0", Long.class, perPartition);
        // 3. 生成分区
        Map<String, ExecutionContext> partitions = new HashMap<>();
        long lastId = Long.MIN_VALUE;
        for (int i = 0; i < boundaries.size(); i++) {
            long maxId = boundaries.get(i);
            ExecutionContext context = new ExecutionContext();
            context.putLong("minId", lastId);
            context.putLong("maxId", maxId);
            partitions.put("partition" + i, context);
            lastId = maxId + 1;
        }
        // 最后一个分区涵盖剩余数据
        if (lastId <= Long.MAX_VALUE) {
            ExecutionContext context = new ExecutionContext();
            context.putLong("minId", lastId);
            context.putLong("maxId", Long.MAX_VALUE);
            partitions.put("partition" + boundaries.size(), context);
        }
        return partitions;
    }
}

注意：大表采样可能较慢，可以预先将统计信息缓存到 Redis 或配置表，或者使用数据库的统计直方图（如 PostgreSQL 的 pg_stats）。此外，避免在 partition 方法中执行昂贵的查询，可以提前在 Job 启动前计算边界并传入参数。

3.3 针对无主键或复合条件的场景

如果分区不能依赖连续 ID（例如需要按 user_id 哈希分区），且某些 user 产生海量数据，需要动态识别热点 user，并将其独立成多个分区，甚至需要二次分区。

热点隔离模式：

1. 先查询 user_id 的订单数，取出 TOP N 大客户。
2. 大客户每个（或几个）独立为一个分区。
3. 其余 user 按 hash 取模均匀分配。

// 伪代码
List<Long> hotUsers = jdbcTemplate.queryForList(
    "SELECT user_id FROM orders GROUP BY user_id HAVING COUNT(*) > ?", Long.class, threshold);
// 为每个热点用户单独分区
hotUsers.forEach(user -> {
    ExecutionContext ctx = new ExecutionContext();
    ctx.putLong("userId", user);
    // 可多个热点合并
    partitions.put("hot-" + user, ctx);
});
// 剩下的用户按 hash 分区，并在 SQL 中排除热点用户

---

四、解决方案二：远程分区 + 动态 Worker 分配，利用多节点弹性均衡

本地分区受限于单机资源，且一旦分区生成，无法在线调整。远程分区（通过消息中间件分发）可以与 Kubernetes 集成，动态扩缩 Worker，但数据倾斜问题依旧存在。可以通过分区再均衡或工作窃取模式解决。

4.1 工作窃取（Work Stealing）模式

将批处理设计为基于消息队列的消费者：分区不再是一次性分配范围，而是生产者将所有待处理的记录的主键列表写入 Redis 队列，多个 Worker 竞争弹出单个 ID 进行处理。这样负载天然均衡。

实现：

• 前置步骤：查询所有需要处理记录的 ID，推送到 Redis List 或 RabbitMQ。
• 每个 Worker 启动一个 ItemReader 从队列中拉取 ID，再根据 ID 加载完整记录进行处理。
• Spring Batch 的 ItemReader 可以这样实现：

public class RedisQueueItemReader implements ItemReader<Long> {
    private RedisTemplate<String, Long> redisTemplate;
    private String queueKey;

    @Override
    public Long read() {
        return redisTemplate.opsForList().rightPop(queueKey); // 阻塞或非阻塞
    }
}

这种方式避免了分区划分的问题，理论上负载绝对均衡。代价是需要额外的前置步骤将数据推入队列，且 ItemReader 需要处理队列空的情况（返回 null 结束 Step）。

4.2 分区后再均衡（Rebalancing）

某些中间件（如 Spring Cloud Data Flow）支持动态分区，在 Worker 间检测负载并重新分配未完成的分区。但其本质也依赖中间件的调度能力。

对于大多数 Spring Batch 项目，工作窃取是避免数据倾斜的最简单有效的终极手段，但它破坏了 Spring Batch 的块处理事务边界（每条记录单独读取，可能无法批处理写入）。解决办法：在 ItemProcessor 或 ItemWriter 中进行微批积累，例如使用 ItemWriter 缓存多条记录后批量写入。

---

五、解决方案三：在数据库层面消除倾斜——数据分片与预聚合

如果倾斜源于数据库本身的数据分布，可以在数据进入批处理之前，利用数据库的分片功能（如分库分表中间件）将数据均匀分布。比如使用 ShardingSphere 按 user_id 哈希分片，每个分片大小均匀。然后 Spring Batch 的分区可以基于分片键（如 user_id 哈希值）等分，天然均衡。

另一种思路是预聚合：将需要处理的记录按某种维度（如时间窗口）预聚合到临时表，并确保聚合后每个窗口的数据量接近。但这会改变业务逻辑，慎用。

5.1 结合 Redis 生成均匀分区边界

在批处理 Job 启动前，使用一个独立的小任务查询数据库，生成均匀的 minId 和 maxId 分段，存储到 Redis 中。然后 Partitioner 从 Redis 读取这些分段，分配给 Worker。这样可以避免在 partition 方法中执行重查询。

---

六、解决方案四：调整 Spring Batch 的并行度与分区分裂

有时倾斜无法完全避免，但可以通过过度分区（gridSize 远大于 Worker 数量）来缓解。例如你只有 4 个 Worker，但将数据切成 64 个分区，然后由 TaskExecutor 并发执行。由于分区更碎，可以更细粒度地混合执行，一个 Worker 完成后会立即拉取下一个分区，从而减少整体等待时间。但要注意不能无限制地增大分区数，因为每个分区有启动开销。

@Bean
public Partitioner partitioner() {
    return new ColumnRangePartitioner() {
        @Override
        public Map<String, ExecutionContext> partition(int gridSize) {
            return super.partition(gridSize * 4); // 夸大分区数
        }
    };
}

配合 TaskExecutorPartitionHandler 的 concurrencyLimit 控制实际并发。

---

七、监控与诊断：及时发现倾斜

7.1 Spring Batch 提供的指标

通过 Micrometer 可收集 spring_batch_step_duration 等指标，比较各个分区的 max 与平均值，若某个分区执行时间远超平均，可告警。

7.2 自定义 StepExecutionListener

在分区 step 中记录每个分区的行数和耗时：

public class PartitionMetricsListener implements StepExecutionListener {
    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        log.info("Partition {} processed {} items in {} ms",
            stepExecution.getStepName(),
            stepExecution.getReadCount(),
            stepExecution.getDuration());
        return null;
    }
}

通过分析日志或推送至监控系统，可视化分区负载分布。

---

八、常见坑点速查表

现象	根因	解决方案
多数分区很快，一个极慢	数据本身按分区键分布不均	改用动态范围分区或热点隔离
按日期分区，某天数据过多	固定范围分区	改用动态采样分区，或将特大日拆分为多个分区
远程分区 Worker 负载不均	静态分区一次性分配	改用消息队列的工作窃取模式
分区启动后发现分布倾斜	未预先分析数据分布	在 Partition 前执行采样查询或使用缓存边界
过度分区导致 OOM	分区数过多，元数据撑爆内存	控制 gridSize，设置合理的 TaskExecutor 队列
分区后每个分区执行仍然很慢	分区内数据处理逻辑有瓶颈（如频繁DB查询）	优化处理逻辑，使用批处理、缓存

---

九、最佳实践：彻底告别“长短腿”分区

1. 先用 SQL 分析数据分布：GROUP BY 关键字段，观察数据密度，作为分区设计依据。
2. 自研动态 Partitioner：依据真实数据量切分范围，而非平均切分 ID。
3. 热点数据特殊对待：大客户、爬虫账号等提前识别并单独分区或拆分。
4. 工作窃取是核武器：当数据倾斜极端严重，放弃静态分区，使用队列模式均分。
5. 过度分区 + 线程池：用更细粒度的分区混合执行，缓解等待。
6. 分区元数据缓存：将分区边界存储于 Redis，避免每次 Job 启动都全表扫描。
7. 监控每个分区耗时：异常倾斜立即通知，并支持自动调整（需人工分析）。
8. 数据库层面的分库分表：从源头解决数据倾斜，是最彻底的方案。

---

十、结语：让每一个分区都“吃饱不撑”

数据倾斜是分布式计算中固有的挑战，在 Spring Batch 中也不例外。它无法被消灭，但可以通过对数据的深入理解、灵活的分区策略和适当的架构调整，被牢牢控制在可接受的范围内。下次设计批处理 Job 时，别再用简单的取模或按日期直接切分了，先跑个 GROUP BY 看看你的数据长什么样，再选择最适合的分区手段。记住，并行化的目的不是让快的分区等慢的分区，而是让所有分区携手撞线。