@Bean
public Job salaryPaymentJob() {
    return jobBuilderFactory.get("salaryPaymentJob")
        .start(step1())
        .next(step2())
        .build();
}

@Bean
public Step salaryPaymentStep() {
    return stepBuilderFactory.get("salaryPaymentStep")
        .<SalaryPayment, SalaryPayment>chunk(1000)
        .reader(reader())
        .processor(processor())
        .writer(writer())
        .build();
}

.faultTolerant()                    // 啟用容錯
    .skipLimit(100)                 // 最多跳過100條
    .skip(IllegalArgumentException.class)    // 跳過數(shù)據(jù)驗(yàn)證異常
    .skip(NullPointerException.class)        // 跳過空指針異常
    .retryLimit(3)                  // 失敗重試3次
    .retry(Exception.class)         // 重試所有異常

@Configuration
public class SalaryPaymentJobConfig {

    @Value("${batch.chunk.size:1000}")
    private int chunkSize;  // 每次處理的記錄數(shù)

    @Value("${batch.skip.limit:100}")
    private int skipLimit;  // 跳過限制

    @Value("${batch.retry.limit:3}")
    private int retryLimit; // 重試次數(shù)

    @Bean
    public Step salaryPaymentStep() {
        return stepBuilderFactory
            .get("salaryPaymentStep")
            .<SalaryPayment, SalaryPayment>chunk(chunkSize)
            .reader(salaryPaymentReader())
            .processor(salaryPaymentProcessor())
            .writer(salaryPaymentWriter())
            .faultTolerant()  // 啟用容錯
            .skipLimit(skipLimit)
            .skip(IllegalArgumentException.class)
            .skip(NullPointerException.class)
            .retryLimit(retryLimit)
            .retry(Exception.class)
            .listener(new SalaryItemReadListener())
            .listener(new SalaryItemWriteListener())
            .build();
    }

    @Bean
    public Job salaryPaymentJob(Step step, SalaryJobExecutionListener listener) {
        return jobBuilderFactory.get("salaryPaymentJob")
            .incrementer(new RunIdIncrementer())
            .listener(listener)
            .start(step)
            .build();
    }
}

@Bean
public FlatFileItemReader<SalaryPayment> salaryPaymentReader() {
    FlatFileItemReader<SalaryPayment> reader = new FlatFileItemReader<>();
    reader.setName("salaryPaymentReader");
    reader.setResource(new ClassPathResource("input/salary-payments.csv"));
    reader.setLinesToSkip(1);  // 跳過CSV標(biāo)題行

    // 設(shè)置列映射
    DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
    tokenizer.setNames(new String[]{
        "employeeId", "employeeName", "accountNumber",
        "accountName", "bankName", "amount", "currency",
        "paymentDate", "remark"
    });

    // 設(shè)置字段映射
    BeanWrapperFieldSetMapper<SalaryPayment> mapper = new BeanWrapperFieldSetMapper<>();
    mapper.setTargetType(SalaryPayment.class);

    DefaultLineMapper<SalaryPayment> lineMapper = new DefaultLineMapper<>();
    lineMapper.setLineTokenizer(tokenizer);
    lineMapper.setFieldSetMapper(mapper);
    reader.setLineMapper(lineMapper);

    return reader;
}

public class SalaryPaymentProcessor implements ItemProcessor<SalaryPayment, SalaryPayment> {

    private static final BigDecimal MIN_AMOUNT = new BigDecimal("0.01");
    private static final BigDecimal MAX_AMOUNT = new BigDecimal("1000000");

    @Override
    public SalaryPayment process(SalaryPayment item) throws Exception {
        // 1. 數(shù)據(jù)驗(yàn)證
        if (item.getEmployeeId() == null || item.getEmployeeId().trim().isEmpty()) {
            throw new IllegalArgumentException("員工ID不能為空");
        }

        // 2. 金額驗(yàn)證
        if (item.getAmount() == null) {
            throw new IllegalArgumentException("發(fā)放金額不能為空");
        }
        if (item.getAmount().compareTo(MIN_AMOUNT) < 0) {
            throw new IllegalArgumentException("發(fā)放金額不能小于0.01元");
        }
        if (item.getAmount().compareTo(MAX_AMOUNT) > 0) {
            throw new IllegalArgumentException("發(fā)放金額不能大于100萬元");
        }

        // 3. 銀行卡號驗(yàn)證
        if (item.getAccountNumber() == null ||
            item.getAccountNumber().length() < 16 ||
            item.getAccountNumber().length() > 19) {
            throw new IllegalArgumentException("銀行賬號長度必須在16-19位之間");
        }

        // 4. 設(shè)置處理狀態(tài)
        item.setStatus("PROCESSING");
        item.setTransactionId("SAL" + System.currentTimeMillis() + item.getEmployeeId());

        return item;
    }
}

public class SalaryPaymentWriter implements ItemWriter<SalaryPayment> {

    private final JdbcBatchItemWriter<SalaryPayment> delegate;

    public SalaryPaymentWriter(DataSource dataSource) {
        this.delegate = new JdbcBatchItemWriter<>();
        this.delegate.setDataSource(dataSource);
        this.delegate.setSql(
            "INSERT INTO salary_payment " +
            "(employee_id, employee_name, account_number, account_name, " +
            "bank_name, amount, currency, payment_date, remark, " +
            "status, transaction_id, create_time, update_time) " +
            "VALUES (:employeeId, :employeeName, :accountNumber, :accountName, " +
            ":bankName, :amount, :currency, :paymentDate, :remark, " +
            ":status, :transactionId, :createTime, :updateTime)");
        this.delegate.setItemSqlParameterSourceProvider(
            new BeanPropertyItemSqlParameterSourceProvider<>()
        );
    }

    @Override
    public void write(List<? extends SalaryPayment> items) throws Exception {
        delegate.write(items);
    }
}

@Component
public class PartialRollbackHandler implements SkipPolicy {

    private static final int SKIP_LIMIT = 100;

    @Override
    public boolean shouldSkip(Throwable throwable, int skipCount) {
        // 超過跳過限制
        if (skipCount >= SKIP_LIMIT) {
            return false;
        }

        // 文件不存在，不能跳過
        if (throwable instanceof FileNotFoundException) {
            return false;
        }

        // 數(shù)據(jù)格式錯誤，可以跳過
        if (throwable instanceof FlatFileParseException) {
            return true;
        }

        // 數(shù)據(jù)驗(yàn)證失敗，可以跳過
        if (throwable instanceof IllegalArgumentException ||
            throwable instanceof NullPointerException) {
            return true;
        }

        return false;
    }
}

CREATE TABLE salary_payment (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    employee_id VARCHAR(50) NOT NULL COMMENT '員工ID',
    employee_name VARCHAR(100) NOT NULL COMMENT '員工姓名',
    account_number VARCHAR(50) NOT NULL COMMENT '銀行賬號',
    account_name VARCHAR(100) NOT NULL COMMENT '賬戶名稱',
    bank_name VARCHAR(100) NOT NULL COMMENT '開戶行',
    amount DECIMAL(18,2) NOT NULL COMMENT '發(fā)放金額',
    currency VARCHAR(10) NOT NULL DEFAULT 'CNY' COMMENT '幣種',
    payment_date DATETIME NOT NULL COMMENT '發(fā)放日期',
    remark VARCHAR(500) COMMENT '備注',
    status VARCHAR(20) NOT NULL DEFAULT 'PENDING' COMMENT '狀態(tài)',
    transaction_id VARCHAR(100) COMMENT '交易ID',
    error_message VARCHAR(1000) COMMENT '錯誤信息',
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_employee_id (employee_id),
    INDEX idx_status (status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

@RestController
@RequestMapping("/api/batch")
public class BatchJobController {

    // 啟動Job
    @PostMapping("/start")
    public ResponseEntity<Map<String, Object>> startJob(
        @RequestParam String inputFile
    ) {
        JobParameters params = new JobParametersBuilder()
            .addLong("startTime", System.currentTimeMillis())
            .addString("inputFile", inputFile)
            .toJobParameters();
        JobExecution execution = jobLauncher.run(salaryPaymentJob, params);
        return ResponseEntity.ok(result);
    }

    // 獲取Job狀態(tài)
    @GetMapping("/status/{jobExecutionId}")
    public ResponseEntity<Map<String, Object>> getJobStatus(
        @PathVariable Long jobExecutionId
    ) {
        JobExecution execution = jobRepository.getJobExecution(jobExecutionId);
        // 返回執(zhí)行詳情
    }

    // 停止Job
    @PostMapping("/stop/{jobExecutionId}")
    public ResponseEntity<Map<String, Object>> stopJob(
        @PathVariable Long jobExecutionId
    ) {
        JobExecution execution = jobRepository.getJobExecution(jobExecutionId);
        execution.stop();
        return ResponseEntity.ok(result);
    }

    // 獲取統(tǒng)計信息
    @GetMapping("/statistics")
    public ResponseEntity<Map<String, Object>> getStatistics() {
        // 返回總數(shù)、成功數(shù)、失敗數(shù)等統(tǒng)計
    }

    // 健康檢查
    @GetMapping("/health")
    public ResponseEntity<Map<String, Object>> health() {
        // 返回系統(tǒng)健康狀態(tài)
    }
}

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("salary-batch-");
    executor.initialize();
    return executor;
}

// 在Step中使用
.step(stepName)
.chunk(chunkSize)
.taskExecutor(taskExecutor())
.throttleLimit(10)  // 限制并發(fā)數(shù)
.build();

@Bean
public Step masterStep() {
    return stepBuilderFactory.get("masterStep")
        .partitioner(slaveStep().getName(), rangePartitioner(1, 10))
        .step(slaveStep())
        .gridSize(10)  // 分成10個分區(qū)
        .taskExecutor(taskExecutor())
        .build();
}

@Bean
public Partitioner rangePartitioner(int min, int max) {
    return new Partitioner() {
        @Override
        public Map<String, ExecutionContext> partition(int gridSize) {
            Map<String, ExecutionContext> result = new HashMap<>();
            int range = (max - min) / gridSize;
            for (int i = 0; i < gridSize; i++) {
                ExecutionContext context = new ExecutionContext();
                context.putInt("minValue", min + i * range);
                context.putInt("maxValue", min + (i + 1) * range - 1);
                result.put("partition" + i, context);
            }
            return result;
        }
    };
}

// 單條插入（慢）
for (SalaryPayment p : payments) {
    jdbcTemplate.update(sql, p.getId(), p.getName(), ...);
}

// 批量插入（快）
jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
    @Override
    public void setValues(PreparedStatement ps, int i) throws SQLException {
        // 設(shè)置參數(shù)
    }
    @Override
    public int getBatchSize() {
        return payments.size();
    }
});

-- 為常用查詢字段添加索引
CREATE INDEX idx_employee_id ON salary_payment(employee_id);
CREATE INDEX idx_status ON salary_payment(status);
CREATE INDEX idx_create_time ON salary_payment(create_time);

-- 復(fù)合索引
CREATE INDEX idx_status_employee ON salary_payment(status, employee_id);

.job(salaryPaymentJob)
    .allowStartIfComplete(false)  // 已完成的Job不重新執(zhí)行
    .restartable(true)  // 允許重啟

@Bean
public Step masterStep() {
    return stepBuilderFactory.get("masterStep")
        .partitioner(slaveStep().getName(), partitioner())
        .step(slaveStep())
        .gridSize(10)  // 分成10個分區(qū)并行處理
        .taskExecutor(taskExecutor())
        .build();
}