Spring Batch多级Chunk实现并行多用户外部数据拉取入库方案咨询

核心改造思路 #

你的核心需求是把原有的用户级Chunk调整为50条数据级Chunk，同时保持多用户并行处理，基于Spring Batch框架可以按以下步骤改造： #

步骤1：改造stepTasklet存储用户列表 #

首先修改第一步的Tasklet逻辑，将加载到的全量用户列表存入JobExecutionContext，供后续并行步骤读取：

@Bean
public Step stepTasklet() {
    return stepBuilderFactory.get("stepTasklet")
            .tasklet((contribution, chunkContext) -> {
                // 原有逻辑：从数据库加载全量用户列表
                List<String> userList = userDao.queryAllUserList();
                // 存入Job上下文，后续步骤可读取
                chunkContext.getStepContext().getJobExecutionContext()
                        .put("fullUserList", userList);
                return RepeatStatus.FINISHED;
            }).build();
}

步骤2：实现线程安全的多用户分页数据Reader #

自定义可线程安全的ItemReader，支持跨用户自动切换、单用户按50条分页拉取外部系统数据：

@Component
@StepScope
public class UserDataReader implements ItemReader<ExternalData> {
    // 全量用户列表
    private List<String> userList;
    // 当前处理的用户索引（线程安全）
    private final AtomicInteger currentUserIndex = new AtomicInteger(0);
    // 当前用户的拉取偏移量（线程安全）
    private final AtomicInteger currentOffset = new AtomicInteger(0);
    // 当前用户已拉取的缓存数据
    private Queue<ExternalData> currentUserDataCache = new ConcurrentLinkedQueue<>();

    @Value("#{jobExecutionContext['fullUserList']}")
    public void setUserList(List<String> userList) {
        this.userList = userList;
    }

    @Override
    public synchronized ExternalData read() throws Exception {
        // 先取缓存里的剩余数据
        if (!currentUserDataCache.isEmpty()) {
            return currentUserDataCache.poll();
        }
        // 缓存为空，检查是否还有未处理的用户
        int userIdx = currentUserIndex.get();
        if (userIdx >= userList.size()) {
            // 所有用户处理完毕
            return null;
        }
        String currentUserId = userList.get(userIdx);
        // 调用外部系统拉取当前用户的50条数据
        List<ExternalData> pageData = externalApi.queryDataByUser(currentUserId, currentOffset.get(), 50);
        if (pageData.isEmpty()) {
            // 当前用户数据拉完，切到下一个用户，重置偏移量
            currentUserIndex.incrementAndGet();
            currentOffset.set(0);
            // 递归读取下一个用户的数据
            return read();
        }
        // 更新偏移量
        currentOffset.addAndGet(pageData.size());
        // 数据存入缓存
        currentUserDataCache.addAll(pageData);
        // 返回第一条数据
        return currentUserDataCache.poll();
    }
}

步骤3：改造parallelStep为数据级Chunk的多线程Step #

调整并行Step为面向Chunk的处理模式，chunk大小设为50，搭配异步线程池实现多用户并行处理：

@Bean
public Step parallelStep(TaskExecutor taskExecutor) {
    return stepBuilderFactory.get("parallelStep")
            .<ExternalData, ExternalData>chunk(50)
            .reader(userDataReader)
            // Processor仅做单条数据校验/转换，无需拉取全量用户数据
            .processor(data -> {
                // 你的单条数据业务处理逻辑
                return data;
            })
            // Writer每次接收50条数据批量写入数据库
            .writer(chunk -> {
                List<ExternalData> dataList = chunk.getItems();
                // 批量写入数据库逻辑
                dataDao.batchInsert(dataList);
            })
            // 配置多线程并行
            .taskExecutor(taskExecutor)
            // 可根据服务器性能调整线程数，建议不超过CPU核心数*2
            .throttleLimit(10)
            .build();
}

// 可选：优化TaskExecutor配置，避免SimpleAsyncTaskExecutor的线程重复创建开销
@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("batch-worker-");
    executor.initialize();
    return executor;
}

步骤4：保留原有Job配置即可 #

你现有Job配置不需要改动，直接复用即可：

@Bean
public Job getParallelJob() {
    JobParametersIncrementer jobParametersIncrementers = new JobParametersIncrementer() {
        long count = new Date().getTime();
        @Override
        public JobParameters getNext( JobParameters parameters) {
            return new JobParameters(Collections.singletonMap("count", new JobParameter(count)));
        }
    };
    return this.jobBuilderFactory.get("paralleljob")
            .incrementer(jobParametersIncrementers)
            .start(stepTasklet())
            .next(parallelStep(taskExecutor()))
            .build();
}

效果验证 #

• 多用户并行：线程池中的多个线程会同时拉取不同用户的数据，不会出现单用户阻塞全量任务的问题
• Chunk粒度为数据级：每次Writer固定接收50条数据批量写入，和用户维度完全解耦，即使单用户有上万条数据也会被拆分为多个50条的Chunk处理
  如果需要更高的并行度，还可以将parallelStep改为分区Step，按用户维度拆分分区，每个分区独立处理一批用户，进一步提升处理效率。

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