Spring Batch 5集成Cassandra批处理需求可行性及替代方案咨询

一、Spring Batch 5 完全可以实现你的需求 #

Spring Batch 5 对 Cassandra 有原生支持，无需传统 JDBC DataSource（Cassandra 是 NoSQL 数据库，依赖 CassandraTemplate 而非 DataSource）。下面是具体实现步骤：

1. 依赖配置 #

在 pom.xml（Maven）或 build.gradle（Gradle）中引入必要依赖：

<!-- Maven 示例 -->
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-batch</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-cassandra</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
</dependencies>

2. Cassandra 配置（替代 AbstractCassandraConfiguration） #

Spring Boot 3+（对应 Spring Batch 5）推荐通过配置文件简化 Cassandra 连接，无需手动继承 AbstractCassandraConfiguration。在 application.properties 中添加：

spring.data.cassandra.contact-points=你的Cassandra地址:9042
spring.data.cassandra.keyspace-name=你的键空间名称
spring.data.cassandra.local-datacenter=你的数据中心名称（如datacenter1）

Spring Boot 会自动配置 CassandraTemplate 和 SessionFactory，供后续批量组件使用。

3. 实现 Cassandra 数据读取器（ItemReader） #

使用 CassandraPagingItemReader 分页读取特定状态的记录：

@Bean
public CassandraPagingItemReader<YourEntity> cassandraReader(CassandraTemplate cassandraTemplate) {
    return new CassandraPagingItemReaderBuilder<YourEntity>()
            .name("cassandraReader")
            .sessionFactory(cassandraTemplate.getSessionFactory())
            // 编写查询特定状态的CQL
            .query(SimpleStatement.builder("SELECT * FROM your_table WHERE status = ?")
                    .addPositionalValue("待处理状态值")
                    .build())
            .pageSize(100) // 每页读取数量，根据数据量调整
            // 映射Cassandra行到实体类
            .rowMapper((row, rowNum) -> {
                YourEntity entity = new YourEntity();
                entity.setId(row.getUUID("id"));
                entity.setStatus(row.getString("status"));
                entity.setNotificationContent(row.getString("notification_content"));
                // 映射其他字段
                return entity;
            })
            .build();
}

4. 实现数据处理器（ItemProcessor） #

在这里完成重发 Kafka 通知、更新状态和时间戳的逻辑：

@Component
public class YourItemProcessor implements ItemProcessor<YourEntity, YourEntity> {

    private final KafkaTemplate<String, String> kafkaTemplate;

    // 构造注入KafkaTemplate
    public YourItemProcessor(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Override
    public YourEntity process(YourEntity item) throws Exception {
        // 重发Kafka通知
        kafkaTemplate.send("你的Kafka主题", item.getId().toString(), item.getNotificationContent());
        // 更新状态与时间戳
        item.setStatus("已重发");
        item.setUpdatedTimestamp(Instant.now());
        return item;
    }
}

5. 实现 Cassandra 数据写入器（ItemWriter） #

将处理后的记录写回 Cassandra：

@Bean
public CassandraItemWriter<YourEntity> cassandraWriter(CassandraTemplate cassandraTemplate) {
    return new CassandraItemWriterBuilder<YourEntity>()
            .sessionFactory(cassandraTemplate.getSessionFactory())
            // 编写更新CQL
            .statementGenerator(item -> SimpleStatement.builder("UPDATE your_table SET status = ?, updated_timestamp = ? WHERE id = ?")
                    .addPositionalValue(item.getStatus())
                    .addPositionalValue(item.getUpdatedTimestamp())
                    .addPositionalValue(item.getId())
                    .build())
            .build();
}

6. 构建 Batch Job 和 Step #

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final CassandraPagingItemReader<YourEntity> cassandraReader;
    private final YourItemProcessor itemProcessor;
    private final CassandraItemWriter<YourEntity> cassandraWriter;

    public BatchConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory,
                       CassandraPagingItemReader<YourEntity> cassandraReader,
                       YourItemProcessor itemProcessor,
                       CassandraItemWriter<YourEntity> cassandraWriter) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.cassandraReader = cassandraReader;
        this.itemProcessor = itemProcessor;
        this.cassandraWriter = cassandraWriter;
    }

    @Bean
    public Step processStep() {
        return stepBuilderFactory.get("processStep")
                .<YourEntity, YourEntity>chunk(100) // 每100条记录提交一次
                .reader(cassandraReader)
                .processor(itemProcessor)
                .writer(cassandraWriter)
                .build();
    }

    @Bean
    public Job processJob() {
        return jobBuilderFactory.get("processJob")
                .start(processStep())
                .build();
    }
}

7. 定时调度配置 #

使用 Spring 自带的 @Scheduled 实现15/30分钟间隔调度：

@Component
@EnableScheduling
public class BatchScheduler {

    private final JobLauncher jobLauncher;
    private final Job processJob;

    public BatchScheduler(JobLauncher jobLauncher, Job processJob) {
        this.jobLauncher = jobLauncher;
        this.processJob = processJob;
    }

    // 每15分钟执行一次，可修改为0 */30 * * * ? 实现30分钟间隔
    @Scheduled(cron = "0 */15 * * * ?")
    public void runBatchJob() throws Exception {
        // 每次执行添加唯一参数，避免Spring Batch拒绝重复执行同一Job实例
        JobParameters jobParameters = new JobParametersBuilder()
                .addLong("executionTime", System.currentTimeMillis())
                .toJobParameters();
        jobLauncher.run(processJob, jobParameters);
    }
}

最后将项目打包为 Jar，部署到服务器后启动即可。

二、替代方案 #

如果不想使用 Spring Batch，可根据数据量和需求复杂度选择以下方案：

• Spring Scheduler + CassandraTemplate：直接在定时任务中查询特定状态记录，循环处理（发 Kafka、更新状态）。优点是代码简单、轻量；缺点是缺乏批量处理的重试、分片、监控等特性，适合数据量较小的场景。
• Apache Flink：适合大数据量、需要流式处理或复杂批量逻辑的场景，原生支持 Cassandra 和 Kafka 集成，也支持定时触发批量任务。缺点是学习成本较高。
• Quartz + CassandraTemplate：用 Quartz 作为调度框架，配合 CassandraTemplate 处理数据。比 Spring Scheduler 支持更灵活的调度配置，但同样没有批量处理的内置特性。

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