自從用了 Spring Batch，效率飆升500%！
場景1：銀行每日利息計算。痛點：?凌晨時段需掃描百萬級賬戶數(shù)據(jù)，手工計算容易遺漏。Spring Batch方案：?分片讀取賬戶數(shù)據(jù)，批量計算利息，失敗自動重試。實際案例：?某銀行系統(tǒng)改造后，利息計算時間從4小時縮短至23分鐘。
一、為什么需要批處理？
1. 應(yīng)用場景解析
場景1：銀行每日利息計算
圖片
- 痛點： 凌晨時段需掃描百萬級賬戶數(shù)據(jù)，手工計算容易遺漏
- Spring Batch方案： 分片讀取賬戶數(shù)據(jù)，批量計算利息，失敗自動重試
- 實際案例： 某銀行系統(tǒng)改造后，利息計算時間從4小時縮短至23分鐘
場景2：電商訂單歸檔
// 傳統(tǒng)SQL示例（存在性能問題）
DELETE FROM active_orders
WHERE create_time < '2023-01-01'
LIMIT 5000; // 需循環(huán)執(zhí)行直到無數(shù)據(jù)- 問題： 直接刪除百萬級數(shù)據(jù)會導(dǎo)致數(shù)據(jù)庫鎖表
- 正確做法： 使用Spring Batch分頁讀取→寫入歷史表→批量刪除
場景3：日志分析
圖片
- 典型需求： 分析Nginx日志中的API響應(yīng)時間分布
- 特殊挑戰(zhàn)： 處理GB級文本文件時的內(nèi)存控制
場景4：醫(yī)療數(shù)據(jù)遷移
圖片
- 特殊要求： 遷移過程中老系統(tǒng)仍需正常使用
- 解決方案： 使用Spring Batch的增量遷移模式
2. 傳統(tǒng)方式痛點
圖片
詳細解釋每個痛點：
- 資源管理復(fù)雜
// 典型的多線程錯誤示例
ExecutorService executor = Executors.newFixedThreadPool(8);
try {
while(hasNextPage()) {
List<Data> page = fetchNextPage();
executor.submit(() -> processPage(page)); // 可能引發(fā)內(nèi)存泄漏
}
} finally {
executor.shutdown(); // 忘記調(diào)用會導(dǎo)致線程堆積
}常見問題：線程池配置不當(dāng)導(dǎo)致OOM、數(shù)據(jù)庫連接泄露
- 容錯性黑洞
// 偽代碼：脆弱的錯誤處理
for (int i=0; i<3; i++) {
try {
processBatch();
break;
} catch (Exception e) {
if (i == 2) sendAlert(); // 簡單重試無法處理部分成功場景
}
}真實案例：某支付系統(tǒng)因未處理部分失敗，導(dǎo)致重復(fù)出款
- 維護噩夢
# 典型硬編碼配置
batch.size=1000
input.path=/data/in
output.path=/data/out問題根源：參數(shù)修改需要重新部署、不同環(huán)境配置混雜
- 監(jiān)控盲區(qū)
# 開發(fā)人員常用的臨時方案
nohup java -jar batch.jar > log.txt 2>&1 &
tail -f log.txt # 無法獲知實時進度關(guān)鍵缺陷：無法回答"處理到哪了？"、"還剩多少？"等業(yè)務(wù)問題
Spring Batch對比優(yōu)勢表
圖片
二、Spring Batch核心架構(gòu)
1. 四大金剛組件深度解析
組件1：Job（作業(yè)工廠）
圖片
- 核心作用： 定義完整的批處理流水線（如月度報表生成流程）
- 真實案例： 某銀行的日終對賬Job包含三個Step
@Bean
public Job reconciliationJob(){
return jobBuilderFactory.get("dailyReconciliation")
.start(downloadBankFileStep())
.next(validateDataStep())
.next(generateReportStep())
.build();
}組件2：Step（裝配流水線）
圖片
設(shè)計模式：采用分塊（Chunk）處理機制
配置示例：
@Bean
public Step importStep(){
return stepBuilderFactory.get("csvImport")
.<User, User>chunk(500) // 每500條提交一次
.reader(csvReader())
.processor(validationProcessor())
.writer(dbWriter())
.faultTolerant()
.skipLimit(10)
.skip(DataIntegrityViolationException.class)
.build();
}組件3：ItemReader（數(shù)據(jù)搬運工）
圖片
- 典型實現(xiàn)：
// 讀取CSV文件示例
@Bean
public FlatFileItemReader<User> csvReader(){
returnnew FlatFileItemReaderBuilder<User>()
.name("userReader")
.resource(new FileSystemResource("data/users.csv"))
.delimited().delimiter(",")
.names("id", "name", "email")
.fieldSetMapper(new BeanWrapperFieldSetMapper<User>() {{
setTargetType(User.class);
}})
.linesToSkip(1) // 跳過標題行
.build();
}組件4：ItemWriter（數(shù)據(jù)收納師）
圖片
- 復(fù)合寫入示例：
@Bean
public CompositeItemWriter<User> compositeWriter(){
returnnew CompositeItemWriterBuilder<User>()
.delegates(dbWriter(), logWriter(), mqWriter())
.build();
}
// 數(shù)據(jù)庫寫入組件
private JdbcBatchItemWriter<User> dbWriter(){
returnnew JdbcBatchItemWriterBuilder<User>()
.dataSource(dataSource)
.sql("INSERT INTO users (name,email) VALUES (:name,:email)")
.beanMapped()
.build();
}2. 架構(gòu)示意圖
圖片
3. 隱藏BOSS：ItemProcessor（數(shù)據(jù)變形金剛）
圖片
- 典型應(yīng)用：數(shù)據(jù)脫敏處理
publicclassDataMaskProcessorimplementsItemProcessor<User, User> {
@Override
public User process(User user){
// 手機號脫敏
String phone = user.getPhone();
user.setPhone(phone.replaceAll("(\\d{3})\\d{4}(\\d{4})", "$1****$2"));
// 郵箱轉(zhuǎn)小寫
user.setEmail(user.getEmail().toLowerCase());
return user;
}
}4. 組件生命周期探秘
圖片
三、手把手開發(fā)指南
1. 環(huán)境搭建
<!-- 完整POM配置 -->
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.1.5</version>
</parent>
<dependencies>
<!-- Batch核心依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<!-- 內(nèi)存數(shù)據(jù)庫（生產(chǎn)環(huán)境可更換為MySQL等） -->
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<scope>runtime</scope>
</dependency>
<!-- Lombok簡化代碼 -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
</dependencies># application.properties
spring.batch.jdbc.initialize-schema=always # 自動創(chuàng)建Batch元數(shù)據(jù)表
spring.datasource.url=jdbc:h2:mem:testdb
spring.datasource.driverClassName=org.h2.Driver2. 第一個批處理任務(wù)
- 領(lǐng)域模型類：
@Data// Lombok注解
@NoArgsConstructor
@AllArgsConstructor
publicclassUser{
private String name;
privateint age;
private String email;
}- 完整Job配置：
@Configuration
@EnableBatchProcessing
publicclassBatchConfig{
@Autowiredprivate JobBuilderFactory jobBuilderFactory;
@Autowiredprivate StepBuilderFactory stepBuilderFactory;
// 定義Job
@Bean
public Job importUserJob(){
return jobBuilderFactory.get("importUserJob")
.start(csvProcessingStep())
.build();
}
// 定義Step
@Bean
public Step csvProcessingStep(){
return stepBuilderFactory.get("csvProcessing")
.<User, User>chunk(100) // 每處理100條提交一次
.reader(userReader())
.processor(userProcessor())
.writer(userWriter())
.build();
}
// CSV文件讀取器
@Bean
public FlatFileItemReader<User> userReader(){
returnnew FlatFileItemReaderBuilder<User>()
.name("userReader")
.resource(new ClassPathResource("users.csv")) // 文件路徑
.delimited()
.delimiter(",")
.names("name", "age", "email") // 字段映射
.targetType(User.class)
.linesToSkip(1) // 跳過標題行
.build();
}
// 數(shù)據(jù)處理（示例：年齡校驗）
@Bean
public ItemProcessor<User, User> userProcessor(){
return user -> {
if (user.getAge() < 0) {
thrownew IllegalArgumentException("年齡不能為負數(shù): " + user);
}
return user.toBuilder() // 使用Builder模式創(chuàng)建新對象
.email(user.getEmail().toLowerCase())
.build();
};
}
// 數(shù)據(jù)庫寫入器
@Bean
public JdbcBatchItemWriter<User> userWriter(DataSource dataSource){
returnnew JdbcBatchItemWriterBuilder<User>()
.dataSource(dataSource)
.sql("INSERT INTO users (name, age, email) VALUES (:name, :age, :email)")
.beanMapped()
.build();
}
}- CSV文件示例（
src/main/resources/users.csv）：
name,age,email
張三,25,zhangsan@example.com
李四,30,lisi@example.com
王五,-5,wangwu@example.com- 啟動類：
@SpringBootApplication
publicclassBatchApplicationimplementsCommandLineRunner{
@Autowired
private JobLauncher jobLauncher;
@Autowired
private Job importUserJob;
publicstaticvoidmain(String[] args){
SpringApplication.run(BatchApplication.class, args);
}
@Override
publicvoidrun(String... args)throws Exception {
JobParameters params = new JobParametersBuilder()
.addLong("startAt", System.currentTimeMillis())
.toJobParameters();
jobLauncher.run(importUserJob, params);
}
}3. 執(zhí)行流程可視化
圖片
4. 運行效果驗證
- 控制臺輸出：
2023-10-01 10:00:00 INFO o.s.b.c.l.support.SimpleJobLauncher - Job: [SimpleJob: [name=importUserJob]] launched
2023-10-01 10:00:05 INFO o.s.batch.core.job.SimpleStepHandler - Executing step: [csvProcessing]
2023-10-01 10:00:15 ERROR o.s.batch.core.step.AbstractStep - Encountered an error executing step csvProcessing
org.springframework.batch.item.validator.ValidationException: 年齡不能為負數(shù): User(name=王五, age=-5, email=wangwu@example.com)- 數(shù)據(jù)庫結(jié)果：
SELECT * FROMusers;
圖片
5. 調(diào)試技巧
- 查看元數(shù)據(jù)：
SELECT * FROM BATCH_JOB_INSTANCE;
SELECT * FROM BATCH_STEP_EXECUTION;- 重試失敗任務(wù)：
// 在Job配置中添加容錯機制
@Bean
public Step csvProcessingStep(){
return stepBuilderFactory.get("csvProcessing")
.<User, User>chunk(100)
.reader(userReader())
.processor(userProcessor())
.writer(userWriter())
.faultTolerant()
.skipLimit(3) // 最多跳過3條錯誤
.skip(IllegalArgumentException.class)
.build();
}- 日志監(jiān)控配置：
logging.level.org.springframework.batch=DEBUG
logging.level.org.hibernate.SQL=WARN四、實戰(zhàn)案例：銀行交易對賬
1. 場景需求增強說明
核心流程：
圖片
技術(shù)挑戰(zhàn)：
- 雙數(shù)據(jù)源讀取（文件+數(shù)據(jù)庫）
- 千萬級數(shù)據(jù)高效比對
- 差異記錄快速入庫
- 分布式環(huán)境運行
2. 完整架構(gòu)設(shè)計
圖片
3. 領(lǐng)域模型定義
@Data
@AllArgsConstructor
@NoArgsConstructor
publicclassTransaction{
// 公共字段
private String transactionId;
private LocalDateTime tradeTime;
private BigDecimal amount;
// 銀行端數(shù)據(jù)
private String bankSerialNo;
private BigDecimal bankAmount;
// 內(nèi)部系統(tǒng)數(shù)據(jù)
private String internalOrderNo;
private BigDecimal systemAmount;
// 對賬結(jié)果
private ReconStatus status;
private String discrepancyType;
}
publicenum ReconStatus {
MATCHED, // 數(shù)據(jù)一致
AMOUNT_DIFF, // 金額不一致
STATUS_DIFF, // 狀態(tài)不一致
ONLY_IN_BANK, // 銀行單邊賬
ONLY_IN_SYSTEM // 系統(tǒng)單邊賬
}4. 完整Job配置
@Configuration
@EnableBatchProcessing
publicclassBankReconJobConfig{
// 主Job定義
@Bean
public Job bankReconciliationJob(Step downloadStep, Step reconStep, Step reportStep){
return jobBuilderFactory.get("bankReconciliationJob")
.start(downloadStep)
.next(reconStep)
.next(reportStep)
.build();
}
// 文件下載Step
@Bean
public Step downloadStep(){
return stepBuilderFactory.get("downloadStep")
.tasklet((contribution, chunkContext) -> {
// 實現(xiàn)SFTP下載邏輯
sftpService.download("/bank/recon/20231001.csv");
return RepeatStatus.FINISHED;
})
.build();
}
// 核心對賬Step
@Bean
public Step reconStep(){
return stepBuilderFactory.get("reconStep")
.<Transaction, Transaction>chunk(1000)
.reader(compositeReader())
.processor(compositeProcessor())
.writer(compositeWriter())
.faultTolerant()
.skipLimit(100)
.skip(DataIntegrityViolationException.class)
.retryLimit(3)
.retry(DeadlockLoserDataAccessException.class)
.build();
}
// 組合數(shù)據(jù)讀取器
@Bean
public CompositeItemReader<Transaction> compositeReader(){
returnnew CompositeItemReaderBuilder<Transaction>()
.delegates(bankFileReader(), internalDbReader())
.build();
}
// 銀行文件讀取器
@Bean
public FlatFileItemReader<Transaction> bankFileReader(){
returnnew FlatFileItemReaderBuilder<Transaction>()
.name("bankFileReader")
.resource(new FileSystemResource("recon/20231001.csv"))
.delimited()
.names("transactionId","tradeTime","amount","bankSerialNo")
.fieldSetMapper(fieldSet -> {
Transaction t = new Transaction();
t.setTransactionId(fieldSet.readString("transactionId"));
t.setBankSerialNo(fieldSet.readString("bankSerialNo"));
t.setBankAmount(fieldSet.readBigDecimal("amount"));
return t;
})
.build();
}
// 內(nèi)部數(shù)據(jù)庫讀取器
@Bean
public JdbcCursorItemReader<Transaction> internalDbReader(){
returnnew JdbcCursorItemReaderBuilder<Transaction>()
.name("internalDbReader")
.dataSource(internalDataSource)
.sql("SELECT order_no, amount, status FROM transactions WHERE trade_date = ?")
.rowMapper((rs, rowNum) -> {
Transaction t = new Transaction();
t.setInternalOrderNo(rs.getString("order_no"));
t.setSystemAmount(rs.getBigDecimal("amount"));
return t;
})
.preparedStatementSetter(ps -> ps.setString(1, "2023-10-01"))
.build();
}
// 組合處理器
@Bean
public CompositeItemProcessor<Transaction> compositeProcessor(){
List<ItemProcessor<?, ?>> delegates = new ArrayList<>();
delegates.add(new DataMatchingProcessor());
delegates.add(new DiscrepancyClassifier());
returnnew CompositeItemProcessorBuilder<>()
.delegates(delegates)
.build();
}
// 組合寫入器
@Bean
public CompositeItemWriter<Transaction> compositeWriter(){
returnnew CompositeItemWriterBuilder<Transaction>()
.delegates(
discrepancyDbWriter(),
alertMessageWriter()
)
.build();
}
}5. 核心處理器實現(xiàn)
publicclassDataMatchingProcessorimplementsItemProcessor<Transaction, Transaction> {
@Override
public Transaction process(Transaction item){
// 雙數(shù)據(jù)源匹配邏輯
if (item.getBankSerialNo() == null) {
item.setStatus(ReconStatus.ONLY_IN_SYSTEM);
} elseif (item.getInternalOrderNo() == null) {
item.setStatus(ReconStatus.ONLY_IN_BANK);
} else {
compareAmounts(item);
compareStatuses(item);
}
return item;
}
privatevoidcompareAmounts(Transaction t){
if (t.getBankAmount().compareTo(t.getSystemAmount()) != 0) {
t.setDiscrepancyType("AMOUNT_MISMATCH");
t.setStatus(ReconStatus.AMOUNT_DIFF);
BigDecimal diff = t.getBankAmount().subtract(t.getSystemAmount());
t.setAmount(diff.abs());
}
}
privatevoidcompareStatuses(Transaction t){
// 假設(shè)從數(shù)據(jù)庫獲取內(nèi)部狀態(tài)
String internalStatus = transactionService.getStatus(t.getInternalOrderNo());
if(!"SETTLED".equals(internalStatus)){
t.setDiscrepancyType("STATUS_MISMATCH");
t.setStatus(ReconStatus.STATUS_DIFF);
}
}
}
publicclassDiscrepancyClassifierimplementsItemProcessor<Transaction, Transaction> {
@Override
public Transaction process(Transaction item){
if (item.getStatus() != ReconStatus.MATCHED) {
// 添加告警標記
item.setAlertLevel(calculateAlertLevel(item));
}
return item;
}
private AlertLevel calculateAlertLevel(Transaction t){
if (t.getAmount().compareTo(new BigDecimal("1000000")) > 0) {
return AlertLevel.CRITICAL;
}
return AlertLevel.WARNING;
}
}6. 差異報告生成Step
@Bean
public Step reportStep(){
return stepBuilderFactory.get("reportStep")
.<Transaction, Transaction>chunk(1000)
.reader(discrepancyReader())
.writer(excelWriter())
.build();
}
@Bean
public JdbcPagingItemReader<Transaction> discrepancyReader(){
returnnew JdbcPagingItemReaderBuilder<Transaction>()
.name("discrepancyReader")
.dataSource(reconDataSource)
.selectClause("SELECT *")
.fromClause("FROM discrepancy_records")
.whereClause("WHERE recon_date = '2023-10-01'")
.sortKeys(Collections.singletonMap("transaction_id", Order.ASCENDING))
.rowMapper(new BeanPropertyRowMapper<>(Transaction.class))
.build();
}
@Bean
public ExcelFileItemWriter<Transaction> excelWriter(){
returnnew ExcelFileItemWriterBuilder<Transaction>()
.name("excelWriter")
.resource(new FileSystemResource("reports/2023-10-01.xlsx"))
.sheetName("差異報告")
.headers(new String[]{"交易ID", "差異類型", "金額差異", "告警級別"})
.fieldExtractor(item -> new Object[]{
item.getTransactionId(),
item.getDiscrepancyType(),
item.getAmount(),
item.getAlertLevel()
})
.build();
}7. 性能優(yōu)化配置
# 應(yīng)用配置
spring.batch.job.enabled=false# 禁止自動啟動
spring.batch.initialize-schema=never # 生產(chǎn)環(huán)境禁止自動建表
# 性能調(diào)優(yōu)參數(shù)
spring.batch.chunk.size=2000 # 根據(jù)內(nèi)存調(diào)整
spring.datasource.hikari.maximum-pool-size=20
spring.jpa.properties.hibernate.jdbc.batch_size=10008. 執(zhí)行監(jiān)控看板
圖片
五、生產(chǎn)級特性
1. 容錯機制
圖片
- 完整容錯配置示例：
@Bean
public Step secureStep(){
return stepBuilderFactory.get("secureStep")
.<Input, Output>chunk(500)
.reader(jdbcReader())
.processor(secureProcessor())
.writer(restApiWriter())
.faultTolerant()
.retryLimit(3)
.retry(ConnectException.class) // 網(wǎng)絡(luò)問題重試
.retry(DeadlockLoserDataAccessException.class) // 數(shù)據(jù)庫死鎖重試
.skipLimit(100)
.skip(DataIntegrityViolationException.class) // 數(shù)據(jù)問題跳過
.skip(InvalidDataAccessApiUsageException.class)
.noRollback(ValidationException.class) // 驗證異常不回滾
.listener(newErrorLogListener()) // 自定義監(jiān)聽器
.build();
}
// 錯誤日志監(jiān)聽器示例
publicclassErrorLogListenerimplementsItemProcessListener<Input, Output> {
@Override
publicvoidonProcessError(Input item, Exception e){
ErrorLog log = new ErrorLog();
log.setItemData(item.toString());
log.setErrorMsg(e.getMessage());
errorLogRepository.save(log);
}
}2. 性能優(yōu)化策略（千萬級數(shù)據(jù)處理）
策略1：并行Step執(zhí)行
圖片
配置代碼：
@Bean
public Job parallelJob(){
return jobBuilderFactory.get("parallelJob")
.start(step1())
.split(new SimpleAsyncTaskExecutor()) // 啟用異步執(zhí)行器
.add(step2(), step3())
.build();
}策略2：分區(qū)處理（Partitioning）
圖片
- 分區(qū)處理器實現(xiàn)：
@Bean
public Step masterStep(){
return stepBuilderFactory.get("masterStep")
.partitioner("slaveStep", partitioner())
.gridSize(10) // 分區(qū)數(shù)量=CPU核心數(shù)*2
.taskExecutor(new ThreadPoolTaskExecutor())
.build();
}
@Bean
public Partitioner partitioner(){
returnnew Partitioner() {
@Override
public Map<String, ExecutionContext> partition(int gridSize){
Map<String, ExecutionContext> result = new HashMap<>();
long total = getTotalRecordCount();
long range = total / gridSize;
for (int i = 0; i < gridSize; i++) {
ExecutionContext context = new ExecutionContext();
context.putLong("min", i * range);
context.putLong("max", (i+1) * range);
result.put("partition"+i, context);
}
return result;
}
};
}
// Slave Step配置
@Bean
public Step slaveStep(){
return stepBuilderFactory.get("slaveStep")
.<Record, Result>chunk(1000)
.reader(rangeReader(null, null))
.processor(processor())
.writer(writer())
.build();
}
@StepScope
@Bean
public ItemReader<Record> rangeReader(
@Value("#{stepExecutionContext[min]}") Long min,
@Value("#{stepExecutionContext[max]}") Long max) {
returnnew JdbcCursorItemReaderBuilder<Record>()
.sql("SELECT * FROM records WHERE id BETWEEN ? AND ?")
.preparedStatementSetter(ps -> {
ps.setLong(1, min);
ps.setLong(2, max);
})
// 其他配置...
.build();
}- 策略3：異步ItemProcessor
圖片
- 異步處理配置：
@Bean
public Step asyncStep(){
return stepBuilderFactory.get("asyncStep")
.<Input, Output>chunk(1000)
.reader(reader())
.processor(asyncItemProcessor())
.writer(writer())
.build();
}
@Bean
public AsyncItemProcessor<Input, Output> asyncItemProcessor(){
AsyncItemProcessor<Input, Output> asyncProcessor = new AsyncItemProcessor<>();
asyncProcessor.setDelegate(syncProcessor()); // 同步處理器
asyncProcessor.setTaskExecutor(new ThreadPoolTaskExecutor());
return asyncProcessor;
}
@Bean
public AsyncItemWriter<Output> asyncItemWriter(){
AsyncItemWriter<Output> asyncWriter = new AsyncItemWriter<>();
asyncWriter.setDelegate(syncWriter()); // 同步寫入器
return asyncWriter;
}3. 性能對比測試數(shù)據(jù)
圖片
優(yōu)化技巧：
- 數(shù)據(jù)庫連接池調(diào)優(yōu)：
spring.datasource.hikari.maximum-pool-size=20
spring.datasource.hikari.minimum-idle=5- JVM參數(shù)優(yōu)化：
java -jar -Xmx4g -XX:+UseG1GC -XX:MaxGCPauseMillis=200 ...- 批處理參數(shù)調(diào)整：
.chunk(2000) // 根據(jù)內(nèi)存容量調(diào)整
.setQueryTimeout(60) // 數(shù)據(jù)庫查詢超時六、監(jiān)控與管理（生產(chǎn)級方案）
1. 監(jiān)控方案升級（Spring Batch Admin替代方案）
圖片
- 現(xiàn)代監(jiān)控棧配置：
// 添加監(jiān)控依賴
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
</dependency>// 暴露監(jiān)控端點
@Bean
public MeterRegistryCustomizer<MeterRegistry> metricsCommonTags(){
return registry -> registry.config().commonTags("application", "batch-service");
}
// 自定義Batch指標
publicclassBatchMetricsListenerextendsJobExecutionListenerSupport{
privatefinal Counter processedRecords = Counter.builder("batch.records.processed")
.description("Total processed records")
.register(Metrics.globalRegistry);
@Override
publicvoidafterStep(StepExecution stepExecution){
processedRecords.increment(stepExecution.getWriteCount());
}
}2. 元數(shù)據(jù)表結(jié)構(gòu)詳解
圖片
關(guān)鍵表用途：
BATCH_JOB_INSTANCE：作業(yè)指紋庫（相同參數(shù)只能存在一個實例）BATCH_JOB_EXECUTION_PARAMS：存儲每次運行的參數(shù)BATCH_STEP_EXECUTION_CONTEXT：保存步驟上下文數(shù)據(jù)（重啟恢復(fù)的關(guān)鍵）
3. 自定義監(jiān)控看板
-- 常用監(jiān)控SQL示例
-- 最近5次作業(yè)執(zhí)行情況
SELECT j.JOB_NAME, e.START_TIME, e.END_TIME,
TIMEDIFF(e.END_TIME, e.START_TIME) ASDURATION,
s.READ_COUNT, s.WRITE_COUNT
FROM BATCH_JOB_EXECUTION e
JOIN BATCH_JOB_INSTANCE j ON e.JOB_INSTANCE_ID = j.JOB_INSTANCE_ID
JOIN BATCH_STEP_EXECUTION s ON e.JOB_EXECUTION_ID = s.JOB_EXECUTION_ID
ORDERBY e.START_TIME DESCLIMIT5;七、常見問題Q&A（終極指南）
1. 內(nèi)存溢出問題深度解決方案
場景：處理10GB CSV文件時OOM
圖片
- 優(yōu)化代碼示例：
@Bean
@StepScope
public FlatFileItemReader<LargeRecord> largeFileReader(
@Value("#{jobParameters['filePath']}") String filePath) {
returnnew FlatFileItemReaderBuilder<LargeRecord>()
.resource(new FileSystemResource(filePath))
.lineMapper(new DefaultLineMapper<>() {{
setLineTokenizer(new DelimitedLineTokenizer());
setFieldSetMapper(new BeanWrapperFieldSetMapper<>() {{
setTargetType(LargeRecord.class);
}});
}})
.linesToSkip(1)
.strict(false) // 允許文件結(jié)尾空行
.saveState(false) // 禁用狀態(tài)保存
.build();
}
// JVM參數(shù)建議
// -XX:+UseG1GC -Xmx2g -XX:MaxGCPauseMillis=2002. 定時任務(wù)高級配置
- 多任務(wù)調(diào)度方案：
@Configuration
@EnableScheduling
publicclassScheduleConfig{
@Autowiredprivate JobLauncher jobLauncher;
@Autowiredprivate Job reportJob;
// 工作日凌晨執(zhí)行
@Scheduled(cron = "0 0 2 * * MON-FRI")
publicvoiddailyJob()throws Exception {
JobParameters params = new JobParametersBuilder()
.addString("date", LocalDate.now().toString())
.toJobParameters();
jobLauncher.run(reportJob, params);
}
// 每小時輪詢
@Scheduled(fixedRate = 3600000)
publicvoidpollJob(){
if(checkNewDataExists()) {
jobLauncher.run(dataProcessJob, new JobParameters());
}
}
// 優(yōu)雅停止示例
publicvoidstopJob(Long executionId){
JobExecution execution = jobExplorer.getJobExecution(executionId);
if(execution.isRunning()) {
execution.setStatus(BatchStatus.STOPPING);
jobRepository.update(execution);
}
}
}3. 高頻問題集錦
Q：如何重新運行失敗的任務(wù)？
-- 步驟1：查詢失敗的任務(wù)ID
SELECT * FROM BATCH_JOB_EXECUTION WHERESTATUS = 'FAILED';
-- 步驟2：使用相同參數(shù)重新啟動
JobParameters params = new JobParametersBuilder()
.addLong("restartId", originalExecutionId)
.toJobParameters();
jobLauncher.run(job, params);Q：處理過程中斷電怎么辦？
圖片
Q：如何實現(xiàn)動態(tài)參數(shù)傳遞？
// 命令行啟動方式
java -jar batch.jar --spring.batch.job.name=dataImportJob date=2023-10-01
// 編程式參數(shù)構(gòu)建
publicvoidrunJobWithParams(Map<String, Object> params){
JobParameters jobParams = new JobParametersBuilder()
.addString("mode", "forceUpdate")
.addLong("timestamp", System.currentTimeMillis())
.toJobParameters();
jobLauncher.run(importJob, jobParams);
}4. 性能調(diào)優(yōu)檢查清單
數(shù)據(jù)庫優(yōu)化
- 添加批量處理索引
- 配置連接池參數(shù)
- 啟用JDBC批處理模式
JVM優(yōu)化
-XX:+UseStringDeduplication
-XX:+UseCompressedOops
-XX:MaxMetaspaceSize=512mBatch配置
spring.batch.jdbc.initialize-schema=never
spring.batch.job.enabled=false
spring.jpa.open-in-view=false責(zé)任編輯：武曉燕
來源：
碼猿技術(shù)專欄
