18. Spring Batch 批处理与定时任务调度

-- 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 的执行上下文

@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();
    }
}

// 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();
}

// 业务转换 + 过滤
@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;
}

@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();
}

// 场景：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，自动从断点恢复

@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;
    }
}

// Worker 端配置：从消息队列接收分区请求
@Bean
public IntegrationFlow workerFlow() {
    return IntegrationFlow.from("requests")
        .handle(stepExecutionRequestHandler())  // 执行 Step
        .channel("replies")
        .get();
}

// 使用 Kafka / RabbitMQ 作为消息中间件分发分区任务

@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);
        }
    }
}

# 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