Spring Boot整合Spring Batch如何实现CouchbaseItemReader

核心选型不要从零实现ItemReader接口，直接继承Spring Batch自带的AbstractPagingItemReader，复用它的分页状态管理、事务边界控制、重启续跑能力。针对百万级数据场景，优先用Couchbase的键集分页（Keyset Pagination） 替代offset分页，避免深分页性能暴跌——毕竟offset到几十万的时候Couchbase扫数据会慢到不可用。

1. 前置依赖准备 #

项目中引入必要的官方依赖即可，不需要额外找第三方包：

<!-- Spring Batch 核心依赖 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<!-- Couchbase 官方Spring Data客户端 -->
<dependency>
    <groupId>org.springframework.data</groupId>
    <artifactId>spring-data-couchbase</artifactId>
</dependency>

2. 核心组件代码实现 #

这个实现支持自定义查询语句、目标实体类型、分页大小、排序字段，默认适配键集分页逻辑，避免深分页性能问题：

import com.couchbase.client.java.query.QueryOptions;
import com.couchbase.client.java.query.QueryScanConsistency;
import org.springframework.batch.item.database.AbstractPagingItemReader;
import org.springframework.data.couchbase.core.CouchbaseTemplate;
import org.springframework.util.Assert;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;

public class CouchbasePagingItemReader<T> extends AbstractPagingItemReader<T> {

    private CouchbaseTemplate couchbaseTemplate;
    // 基础查询语句，末尾不需要手动拼接limit/offset，组件自动处理
    private String baseQuery;
    // 查询结果映射的目标实体类型
    private Class<T> targetType;
    // 键集分页用的排序字段，必须建索引，默认用文档ID
    private String sortField = "META().id";
    // 上一页最后一条记录的排序字段值，用于键集分页过滤
    private Object lastSortValue;
    // N1QL查询绑定参数
    private Map<String, Object> queryParams = new HashMap<>();

    public CouchbasePagingItemReader() {
        setName("couchbasePagingItemReader");
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        super.afterPropertiesSet();
        Assert.notNull(couchbaseTemplate, "CouchbaseTemplate 不能为null");
        Assert.hasText(baseQuery, "基础查询语句不能为空");
        Assert.notNull(targetType, "目标映射类型不能为null");
        // 强制校验排序规则，避免分页结果乱序、漏读
        Assert.isTrue(baseQuery.toUpperCase().contains("ORDER BY"), "查询语句必须显式指定ORDER BY规则，且排序字段需和配置的sortField一致");
    }

    @Override
    @SuppressWarnings("unchecked")
    protected void doReadPage() {
        if (results == null) {
            results = new CopyOnWriteArrayList<>();
        } else {
            results.clear();
        }

        // 拼接最终查询语句：第一页直接查询，后续用键值过滤，避免深分页性能问题
        StringBuilder finalQuery = new StringBuilder(baseQuery);
        Map<String, Object> bindParams = new HashMap<>(queryParams);
        if (lastSortValue != null) {
            finalQuery.append(" AND ").append(sortField).append(" > $lastSortValue");
            bindParams.put("lastSortValue", lastSortValue);
        }
        finalQuery.append(" LIMIT $pageSize");
        bindParams.put("pageSize", getPageSize());

        // 执行查询，离线批处理场景用NOT_BOUNDED一致性，读性能最高
        List<T> pageResult = couchbaseTemplate.findByQuery(targetType)
                .withOptions(QueryOptions.queryOptions()
                        .scanConsistency(QueryScanConsistency.NOT_BOUNDED)
                        .parameters(com.couchbase.client.java.query.QueryParameters.from(bindParams)))
                .all();

        results.addAll(pageResult);

        // 更新排序游标值，供下一页查询使用
        if (!pageResult.isEmpty()) {
            T lastItem = pageResult.get(pageResult.size() - 1);
            if ("META().id".equals(sortField)) {
                lastSortValue = couchbaseTemplate.getBeanInformation(lastItem).getRequiredId();
            } else {
                // 反射取实体对应字段值，生产环境可缓存反射对象优化性能
                try {
                    var field = targetType.getDeclaredField(sortField);
                    field.setAccessible(true);
                    lastSortValue = field.get(lastItem);
                } catch (Exception e) {
                    throw new RuntimeException("读取排序字段值失败", e);
                }
            }
        }
    }

    @Override
    protected void doJumpToPage(int itemIndex) {
        // 若需要支持任意位置断点重启，只需把lastSortValue存入执行上下文，重启时预查对应位置的排序值赋值即可
    }

    // 省略各字段的setter方法，可根据项目习惯补充或用Builder模式构建
    public void setCouchbaseTemplate(CouchbaseTemplate couchbaseTemplate) { this.couchbaseTemplate = couchbaseTemplate; }
    public void setBaseQuery(String baseQuery) { this.baseQuery = baseQuery; }
    public void setTargetType(Class<T> targetType) { this.targetType = targetType; }
    public void setSortField(String sortField) { this.sortField = sortField; }
    public void setQueryParams(Map<String, Object> queryParams) { this.queryParams = queryParams; }
}

3. 百万级数据场景优化注意点 #

• 页大小不要设太大，建议1000-5000，平衡查询次数和单条查询的内存占用，避免单条返回数据太多撑爆JVM
• 排序字段必须建覆盖索引，比如你要查的字段是userId、orderAmount、createTime，排序用createTime+META().id，那索引就要把这几个字段全包含，避免回表查询，百万级数据下查询速度能提10倍以上
• 批量任务读的时候不要用REQUEST_PLUS一致性，除非你强要求读到任务启动时刻的全量最新数据，不然NOT_BOUNDED一致性的读吞吐量高30%以上，离线批处理场景完全够用
• 不要在Reader里做任何数据加工逻辑，Reader只负责读，加工逻辑全放ItemProcessor，写逻辑用批量Writer，比如Couchbase的批量upsert接口，减少网络IO
• 如果数据量超过千万，直接换Couchbase Analytics服务用游标读，比N1QL分页更稳，代码里只要把doReadPage里的查询换成Analytics查询即可，核心逻辑不变

4. 实际使用示例 #

在Spring Batch的配置类中直接声明Reader、Step、Job即可：

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.couchbase.core.CouchbaseTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class BatchJobConfig {

    @Bean
    public CouchbasePagingItemReader<OrderDTO> orderItemReader(CouchbaseTemplate couchbaseTemplate) {
        CouchbasePagingItemReader<OrderDTO> reader = new CouchbasePagingItemReader<>();
        reader.setCouchbaseTemplate(couchbaseTemplate);
        reader.setTargetType(OrderDTO.class);
        reader.setPageSize(2000);
        reader.setSortField("createTime");
        // 基础查询语句：注意必须显式写ORDER BY规则
        reader.setBaseQuery("SELECT META().id, userId, orderAmount, createTime FROM order_bucket WHERE docType = 'order' AND createTime < $targetTime ORDER BY createTime ASC, META().id ASC");
        Map<String, Object> params = new HashMap<>();
        params.put("targetTime", 1704067200000L);
        reader.setQueryParams(params);
        return reader;
    }

    @Bean
    public Step orderProcessStep(JobRepository jobRepository, PlatformTransactionManager transactionManager,
                                 CouchbasePagingItemReader<OrderDTO> reader,
                                 OrderItemProcessor processor,
                                 OrderItemWriter writer) {
        return new StepBuilder("orderProcessStep", jobRepository)
                .<OrderDTO, ProcessedOrderDTO>chunk(2000, transactionManager)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .build();
    }

    @Bean
    public Job orderProcessJob(JobRepository jobRepository, Step orderProcessStep) {
        return new JobBuilder("orderProcessJob", jobRepository)
                .start(orderProcessStep)
                .build();
    }
}

踩坑提醒：绝对不要用offset+limit的方式做分页，我之前线上踩过坑，页大小2000的情况下，读到第500页的时候单条查询耗时就从10ms涨到了2s，越往后越慢。键集分页不管翻到多少页，查询耗时基本稳定在10ms上下，跑百万级数据总耗时能差十几倍。另外记得给Reader设置saveState=true，把lastSortValue存到Spring Batch的执行上下文里，任务中断重启的时候能从上次读到的位置继续，不用从头读。

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