企业级应用中，大量场景需要批量处理数据：每日对账、月度结算、历史数据迁移、报表生成等。Spring Batch 是 Java 生态中最成熟的批处理框架，提供了作业编排、事务管理、容错重试、进度监控等完整能力。
1. Spring Batch 核心架构
1.1 三层模型
Job（作业）
├── JobInstance（一次运行实例，由 JobParameters 区分）
│ └── JobExecution（实际执行状态）
│
└── Step（步骤）
├── StepExecution
│
└── 三种处理模式:
1. Tasklet（单任务，如清理、通知）
2. Chunk-Oriented（读→处理→写，主流模式）
3. Partition（分区并行处理）
1.2 核心组件
| 组件 | 职责 |
|---|---|
JobRepository | 持久化 Job/Step 执行元数据（数据库表） |
JobLauncher | 启动作业的入口 |
JobExplorer | 查询历史执行状态 |
ItemReader | 从数据源读取数据 |
ItemProcessor | 业务转换/过滤 |
ItemWriter | 写入目标存储 |
1.3 批处理元数据库表
-- Spring Batch 自动创建以下核心表
BATCH_JOB_INSTANCE -- Job 实例（相同的 Job + Parameters = 同一 Instance）
BATCH_JOB_EXECUTION -- 每次执行记录
BATCH_JOB_EXECUTION_PARAMS -- 执行参数
BATCH_STEP_EXECUTION -- Step 执行记录
BATCH_JOB_EXECUTION_CONTEXT -- 跨 Step 共享的执行上下文
BATCH_STEP_EXECUTION_CONTEXT -- 单个 Step 的执行上下文
2. Chunk-Oriented 处理
2.1 Chunk 流程
ItemReader ItemProcessor ItemWriter
│ │ │
↓ ↓ ↓
read() ──→ List<Item> ──→ process() ──→ List<Item> ──→ write()
↑ ↑
└──── 达到 commit-interval ──────┘
事务边界: 每 commit-interval 个 item 构成一个事务单元，失败则回滚整批。
2.2 基础 Job 定义
@Configuration
public class DailyReportJobConfig {
@Bean
public Job dailyReportJob(JobRepository jobRepository,
Step reportStep) {
return new JobBuilder("dailyReportJob", jobRepository)
.incrementer(new RunIdIncrementer()) // 每次运行参数+1，允许重复执行
.start(reportStep)
.build();
}
@Bean
public Step reportStep(JobRepository jobRepository,
PlatformTransactionManager transactionManager,
ItemReader<Order> orderReader,
ItemProcessor<Order, ReportDTO> reportProcessor,
ItemWriter<ReportDTO> reportWriter) {
return new StepBuilder("reportStep", jobRepository)
.<Order, ReportDTO>chunk(100, transactionManager) // 每 100 条提交一次
.reader(orderReader)
.processor(reportProcessor)
.writer(reportWriter)
.faultTolerant() // 启用容错
.skipLimit(10) // 最多跳过 10 条异常
.skip(DataIntegrityException.class)
.retryLimit(3) // 写失败重试 3 次
.retry(TransientDataAccessException.class)
.listener(new StepExecutionListener() {
@Override
public void beforeStep(StepExecution stepExecution) {
log.info("开始处理: {}", stepExecution.getStepName());
}
@Override
public ExitStatus afterStep(StepExecution stepExecution) {
log.info("处理完成: 读取={}, 写出={}",
stepExecution.getReadCount(),
stepExecution.getWriteCount());
return ExitStatus.COMPLETED;
}
})
.build();
}
}
2.3 ItemReader 实现
// JDBC 分页读取（处理大数据量，避免内存溢出）
@Bean
public JdbcPagingItemReader<Order> orderReader(DataSource dataSource) {
return new JdbcPagingItemReaderBuilder<Order>()
.name("orderReader")
.dataSource(dataSource)
.rowMapper(new OrderRowMapper())
.queryProvider(new SqlPagingQueryProviderFactoryBean() {{
setSelectClause("SELECT order_id, user_id, amount, status, create_time");
setFromClause("FROM t_order");
setWhereClause("create_time >= :startDate AND create_time < :endDate");
setSortKey("order_id");
}})
.pageSize(100)
.parameterValues(Map.of(
"startDate", LocalDate.now().minusDays(1).atStartOfDay(),
"endDate", LocalDate.now().atStartOfDay()
))
.build();
}
// Flat File 读取（CSV）
@Bean
public FlatFileItemReader<CsvRecord> csvReader() {
return new FlatFileItemReaderBuilder<CsvRecord>()
.name("csvReader")
.resource(new FileSystemResource("input/data.csv"))
.linesToSkip(1) // 跳过表头
.delimited()
.names("id", "name", "amount")
.fieldSetMapper(new BeanWrapperFieldSetMapper<>() {{
setTargetType(CsvRecord.class);
}})
.build();
}
// JPA 读取
@Bean
public JpaPagingItemReader<Order> jpaOrderReader(EntityManagerFactory emf) {
return new JpaPagingItemReaderBuilder<Order>()
.name("jpaOrderReader")
.entityManagerFactory(emf)
.queryString("SELECT o FROM Order o WHERE o.status = :status")
.parameterValues(Map.of("status", OrderStatus.PENDING))
.pageSize(50)
.build();
}
2.4 ItemProcessor 与 ItemWriter
// 业务转换 + 过滤
@Bean
public ItemProcessor<Order, ReportDTO> reportProcessor() {
return order -> {
if (order.getAmount().compareTo(BigDecimal.ZERO) <= 0) {
return null; // 返回 null 表示过滤掉
}
return ReportDTO.builder()
.orderId(order.getId())
.amount(order.getAmount())
.category(classifyCategory(order))
.reportDate(LocalDate.now())
.build();
};
}
// 写入数据库
@Bean
public JdbcBatchItemWriter<ReportDTO> reportWriter(DataSource dataSource) {
return new JdbcBatchItemWriterBuilder<ReportDTO>()
.itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
.sql("INSERT INTO t_daily_report (order_id, amount, category, report_date) " +
"VALUES (:orderId, :amount, :category, :reportDate)")
.dataSource(dataSource)
.build();
}
// 组合处理器（链式）
@Bean
public ItemProcessor<Order, ReportDTO> compositeProcessor() {
CompositeItemProcessor<Order, ReportDTO> processor = new CompositeItemProcessor<>();
processor.setDelegates(Arrays.asList(
new ValidationProcessor(),
new AggregationProcessor(),
new EnrichmentProcessor()
));
return processor;
}
3. 容错与重试
3.1 Skip 策略
@Bean
public Step robustStep(JobRepository jr, PlatformTransactionManager tm) {
return new StepBuilder("robustStep", jr)
.<Order, Order>chunk(50, tm)
.reader(reader())
.processor(processor())
.writer(writer())
.faultTolerant()
// 跳过配置
.skipLimit(100) // 总共最多跳过 100 条
.skip(DataIntegrityViolationException.class)
.skip(ValidationException.class)
.noSkip(FileNotFoundException.class) // Fatal 异常不跳过
// 重试配置
.retryLimit(3)
.retry(DeadlockLoserDataAccessException.class)
.retry(TransientDataAccessException.class)
// 跳过监听：记录到错误表
.listener(new SkipListener<Order, Order>() {
@Override
public void onSkipInWrite(Order item, Throwable t) {
errorLogRepository.save(new ErrorLog(item.getId(), t.getMessage()));
}
})
.build();
}
3.2 Restart 机制
// 场景：Job 执行到 50 万条时失败，重启后从断点继续
// Job 必须配置为可重启
@Bean
public Job restartableJob(JobRepository jobRepository, Step step) {
return new JobBuilder("dataMigrationJob", jobRepository)
.incrementer(new RunIdIncrementer())
.preventRestart() // ❌ 设置后不可重启
.start(step)
.build();
}
// 查询上次执行状态
@Autowired private JobExplorer jobExplorer;
public JobExecution getLastExecution(String jobName) {
List<JobInstance> instances = jobExplorer.findJobInstancesByJobName(jobName, 0, 1);
if (instances.isEmpty()) return null;
List<JobExecution> executions = jobExplorer.getJobExecutions(instances.get(0));
return executions.get(0); // 最新的执行
}
// RESTART 执行（使用相同 JobParameters）
JobParameters params = new JobParametersBuilder()
.addString("date", "2024-01-28")
.toJobParameters();
jobLauncher.run(job, params); // 如果上次 FAILED，自动从断点恢复
4. 分区并行处理
4.1 Partition Step 架构
Master Step (PartitionHandler)
├── Worker Step 1: 处理 id 1-10000
├── Worker Step 2: 处理 id 10001-20000
├── Worker Step 3: 处理 id 20001-30000
└── Worker Step 4: 处理 id 30001-40000
每个 Worker 在独立线程中执行
4.2 基于范围的分区
@Configuration
public class PartitionJobConfig {
@Bean
public Step masterStep(JobRepository jr,
PartitionHandler partitionHandler) {
return new StepBuilder("masterStep", jr)
.partitioner("workerStep", rangePartitioner()) // 定义分区策略
.step(workerStep(null, null, null, null))
.partitionHandler(partitionHandler)
.build();
}
@Bean
public Partitioner rangePartitioner() {
return gridSize -> {
Map<String, ExecutionContext> partitions = new HashMap<>();
int minId = 1;
int maxId = 1000000;
int range = (maxId - minId) / gridSize; // gridSize = 4
for (int i = 0; i < gridSize; i++) {
ExecutionContext context = new ExecutionContext();
int start = minId + (i * range);
int end = (i == gridSize - 1) ? maxId : start + range - 1;
context.putInt("minId", start);
context.putInt("maxId", end);
partitions.put("partition" + i, context);
}
return partitions;
};
}
@Bean
public Step workerStep(JobRepository jr,
PlatformTransactionManager tm,
@Qualifier("partitionReader") ItemReader<Order> reader,
ItemProcessor<Order, Order> processor,
ItemWriter<Order> writer) {
return new StepBuilder("workerStep", jr)
.<Order, Order>chunk(100, tm)
.reader(reader)
.processor(processor)
.writer(writer)
.build();
}
@Bean
@StepScope // 每个分区创建新实例，注入分区参数
public JdbcPagingItemReader<Order> partitionReader(
@Value("#{stepExecutionContext['minId']}") int minId,
@Value("#{stepExecutionContext['maxId']}") int maxId,
DataSource dataSource) {
return new JdbcPagingItemReaderBuilder<Order>()
.name("partitionReader")
.dataSource(dataSource)
.rowMapper(new OrderRowMapper())
.queryProvider(new SqlPagingQueryProviderFactoryBean() {{
setSelectClause("SELECT *");
setFromClause("FROM t_order");
setWhereClause("id BETWEEN " + minId + " AND " + maxId);
setSortKey("id");
}})
.pageSize(100)
.build();
}
@Bean
public TaskExecutorPartitionHandler partitionHandler(
TaskExecutor taskExecutor) {
TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
handler.setTaskExecutor(taskExecutor);
handler.setStep(workerStep(null, null, null, null, null));
handler.setGridSize(4); // 4 个分区
return handler;
}
}
4.3 远程分区（Remote Partitioning）
Spring Batch Integration 支持将 Worker 分发到多个 JVM 实例：
// Worker 端配置：从消息队列接收分区请求
@Bean
public IntegrationFlow workerFlow() {
return IntegrationFlow.from("requests")
.handle(stepExecutionRequestHandler()) // 执行 Step
.channel("replies")
.get();
}
// 使用 Kafka / RabbitMQ 作为消息中间件分发分区任务
5. Quartz 定时调度整合
5.1 Spring Batch + Quartz 架构
Quartz Scheduler
├── Trigger (Cron: 0 0 2 * * ? 每天2点)
│ └── JobDetail (DailyReportQuartzJob)
│ └── execute():
│ JobParameters params = ...
│ jobLauncher.run(springBatchJob, params)
5.2 Quartz 配置
@Configuration
public class QuartzConfig {
@Bean
public JobDetail dailyReportJobDetail() {
return JobBuilder.newJob(BatchJobLauncher.class)
.withIdentity("dailyReportJob")
.usingJobData("jobName", "dailyReportJob") // 传递 Spring Batch Job 名
.storeDurably()
.build();
}
@Bean
public Trigger dailyReportTrigger() {
// 每天凌晨 2 点执行
CronScheduleBuilder schedule = CronScheduleBuilder
.dailyAtHourAndMinute(2, 0)
.inTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
return TriggerBuilder.newTrigger()
.forJob(dailyReportJobDetail())
.withIdentity("dailyReportTrigger")
.withSchedule(schedule)
.build();
}
}
// Quartz Job 实现：桥接到 Spring Batch
public class BatchJobLauncher implements Job {
@Autowired private JobLauncher jobLauncher;
@Autowired private JobLocator jobLocator;
@Override
public void execute(JobExecutionContext context) throws JobExecutionException {
String jobName = context.getMergedJobDataMap().getString("jobName");
try {
Job job = jobLocator.getJob(jobName);
JobParameters params = new JobParametersBuilder()
.addDate("runTime", new Date())
.addString("date", LocalDate.now().toString())
.toJobParameters();
JobExecution execution = jobLauncher.run(job, params);
if (!execution.getStatus().isUnsuccessful()) {
throw new JobExecutionException("Job failed: " + execution.getStatus());
}
} catch (Exception e) {
throw new JobExecutionException(e);
}
}
}
5.3 集群调度（Quartz JDBC JobStore）
# quartz.properties
org.quartz.scheduler.instanceName = ClusteredScheduler
org.quartz.scheduler.instanceId = AUTO
# 使用数据库存储，支持集群
org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX
org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.StdJDBCDelegate
org.quartz.jobStore.dataSource = myDS
org.quartz.jobStore.tablePrefix = QRTZ_
org.quartz.jobStore.isClustered = true
org.quartz.jobStore.clusterCheckinInterval = 20000
-- Quartz 集群表（11张）
QRTZ_JOB_DETAILS, QRTZ_TRIGGERS, QRTZ_SIMPLE_TRIGGERS
QRTZ_CRON_TRIGGERS, QRTZ_SIMPROP_TRIGGERS
QRTZ_BLOB_TRIGGERS, QRTZ_CALENDARS
QRTZ_PAUSED_TRIGGER_GRPS, QRTZ_SCHEDULER_STATE
QRTZ_LOCKS, QRTZ_FIRED_TRIGGERS
6. 性能优化建议
| 优化点 | 建议 |
|---|---|
| Chunk Size | 100-1000，根据内存和 DB 性能调整 |
| 分页读取 | 大数据量使用 JdbcPagingItemReader |
| 批量写入 | JdbcBatchItemWriter 开启批量模式 |
| 无状态 | ItemReader/Writer 应该是无状态的（@StepScope 除外） |
| 线程池 | 分区时 TaskExecutor 核心线程数 = 分区数 |
| 事务隔离 | READ_COMMITTED 足以，避免 SERIALIZABLE |
7. Step Flow 条件分支与决策路由
除了线性执行 stepA → stepB → stepC，Spring Batch 支持基于执行状态的条件路由：
@Bean
public Job conditionalJob(JobRepository jr,
Step validateStep, Step processStep,
Step retryStep, Step notifyStep) {
return new JobBuilder("conditionalJob", jr)
.start(validateStep)
.on("COMPLETED").to(processStep) // 校验通过 → 处理
.from(validateStep).on("FAILED").to(retryStep) // 校验失败 → 重试
.from(retryStep).on("COMPLETED").to(processStep) // 重试成功 → 继续
.from(retryStep).on("FAILED").to(notifyStep) // 重试失败 → 告警
.from(processStep).on("*").to(notifyStep) // 无论成败都通知
.end()
.build();
}
更复杂的业务决策可自定义 JobExecutionDecider，基于运行时数据（如文件大小、日期奇偶性）决定路由：
public class FileSizeDecider implements JobExecutionDecider {
@Override
public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
Long fileSize = jobExecution.getExecutionContext().getLong("fileSize");
return fileSize > 1024 * 1024 * 100
? new FlowExecutionStatus("LARGE")
: new FlowExecutionStatus("SMALL");
}
}
// 使用决策器
.start(processStep)
.next(new FileSizeDecider())
.on("LARGE").to(partitionStep)
.from(new FileSizeDecider()).on("SMALL").to(simpleStep)
8. 监听器体系与可观测性
Spring Batch 提供了六类监听器，覆盖 Job/Step/Chunk/Item 全生命周期：
| 监听器 | 触发时机 | 典型用途 |
|---|---|---|
JobExecutionListener | Job 开始/结束 | 发送开始通知、记录总耗时 |
StepExecutionListener | Step 开始/结束 | Step 级日志、指标上报 |
ChunkListener | Chunk 前后/onError | 事务边界日志、Chunk 失败熔断 |
ItemReadListener | 每条读取前后 | 读取速率监控、异常数据标记 |
ItemProcessListener | 处理前后/onError | 业务校验失败统计 |
ItemWriteListener | 写入前后/onError | 写入延迟监控、批量写入审计 |
通过 Micrometer 集成，批处理作业可自动暴露结构化指标到 Prometheus：
@Configuration
public class BatchMetricsConfig {
@Bean
public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor(
JobRegistry jobRegistry) {
JobRegistryBeanPostProcessor postProcessor = new JobRegistryBeanPostProcessor();
postProcessor.setJobRegistry(jobRegistry);
return postProcessor;
}
}
Spring Boot 3 + Batch 5 默认集成 Micrometer，关键指标包括：
| 指标名 | 类型 | 含义 |
|---|---|---|
spring.batch.job | Timer | Job 执行耗时（含标签：name, status） |
spring.batch.step | Timer | Step 执行耗时 |
spring.batch.item.read | Counter | 读取条数 |
spring.batch.item.write | Counter | 写入条数 |
spring.batch.chunk.errors | Counter | Chunk 级错误数 |
建议将 BATCH_JOB_EXECUTION 表关键字段同步到 Grafana，构建批处理运营看板：最后一次成功时间、当天执行总量、失败率趋势。
9. 消息队列整合：事件驱动批处理
对于上游系统不断产生数据的场景，可用消息队列替代定时拉取：
// Kafka 作为 ItemReader —— 消费分区数据后提交 offset
@Bean
public KafkaItemReader<String, OrderEvent> kafkaReader(
ConsumerFactory<String, OrderEvent> consumerFactory) {
Properties props = new Properties();
props.putAll(consumerFactory.getConfigurationProperties());
return new KafkaItemReaderBuilder<String, OrderEvent>()
.name("kafkaOrderReader")
.consumerProperties(props)
.topic("order-events")
.partitions(0, 1, 2, 3) // 分配的 partition
. pollTimeout(Duration.ofSeconds(5))
.saveState(true) // 失败重启后从上次 offset 恢复
.build();
}
// 写入 RabbitMQ 以实现下游系统解耦
@Bean
public AmqpItemWriter<ProcessedOrder> amqpWriter(AmqpTemplate amqpTemplate) {
return new AmqpItemWriterBuilder<ProcessedOrder>()
.amqpTemplate(amqpTemplate)
.routingKey("processed.orders")
.build();
}
| 场景 | 方案 | 优势 |
|---|---|---|
| 上游持续数据流入 | KafkaItemReader | 实时消费，失败可重放 |
| 下游多系统分发 | AmqpItemWriter | 路由灵活，解耦处理 |
| 定时大批量处理 | JDBC/FlatFile Reader | 吞吐高，控制精确 |
消息驱动批处理适合近实时场景（延迟秒级），而定时调度适合日终/月终等批量结算场景。两者可在同一系统中并存：Kafka Reader 处理实时订单，Quartz 调度处理日终报表。
10. 实战补充：银行日终对账批处理
对账是批处理的典型高价值场景，其编排流程如下：
Job: dailyReconciliationJob
├── Step 1: downloadFiles (Tasklet) —— 下载银行与核心系统文件
├── Step 2: validateFormat (Chunk) —— 校验文件格式与完整性
├── Step 3: loadBankData (Chunk) —— 加载银行侧交易流水
├── Step 4: loadCoreData (Chunk) —— 加载核心系统交易流水
├── Step 5: matchTransactions (Chunk) —— 按交易号匹配，标记 matched/unmatched
├── Step 6: generateReport (Chunk) —— 生成差异报表
└── Step 7: sendNotification (Tasklet) —— 邮件通知运营团队
对账批处理的核心挑战是幂等性。交易号唯一索引 + 先查后插机制确保同一文件重复处理不会生成重复差异记录。RunIdIncrementer 会导致每次执行参数不同，对账 Job 应使用业务日期作为参数，配合 JobExplorer 检查当天是否已执行成功，避免重复处理：
public JobParameters createReconParams(LocalDate reconDate) {
return new JobParametersBuilder()
.addString("reconDate", reconDate.toString())
.addLong("timestamp", System.currentTimeMillis()) // 强制唯一，防止参数相同
.toJobParameters();
}
延伸阅读
- Java 容器化与 K8s 部署 — K8s CronJob 作为定时调度替代方案
- 监控诊断与可观测性 — 批处理作业监控
- Java 异步与响应式编程 — 与异步处理的对比选型
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。