基于指定Cassandra表结构，Java线程优化批量取数与处理的技术问询

阿华AIGC实验室

2026-5-20

如何用Java高效批量获取Cassandra记录并多线程处理

刚好做过类似的Cassandra批量处理场景，结合你的表结构给你梳理一套最优性能的方案，亲测好用！

一、先摸清楚你的表结构逻辑

你的两个表设计很贴合Cassandra的分区思路：

PRODUCT_INV用(Product_Code, Storecode, StoreType)做复合分区键，能快速定位单个产品在特定门店的库存；
PRODUCT_BY_STORE则按(Storecode, StoreType)聚合该门店下的所有产品编码，方便按门店维度批量拉取库存数据。

接下来的方案就围绕这两个表的特性展开。

二、批量拉取数据的最优姿势

1. 用好Cassandra的分页查询，拒绝一次性拉全量

Cassandra默认就会分页返回结果，千万别想着把所有数据一次性拉到内存里——数据量大的话直接OOM。你可以通过setFetchSize()调整每次从Cassandra拉取的行数，一般1000-5000是比较稳妥的范围，根据你的JVM内存大小调整：

Session session = cluster.connect();
// 单条件查询示例，适配PRODUCT_INV的分区键
Statement stmt = new SimpleStatement("SELECT * FROM PRODUCT_INV WHERE Product_Code = ? AND Storecode = ? AND StoreType = ?")
        .setFetchSize(2000); // 按需调整
ResultSet rs = session.execute(stmt.bind("12", "601", "Retail"));
for (Row row : rs) {
    // 这里可以直接处理，或者先丢到线程安全的队列里给后续线程处理
}

2. 结合PRODUCT_BY_STORE做批量查询（重点！）

如果要批量获取某个门店下所有产品的库存，先从PRODUCT_BY_STORE拿到产品编码集合，再拆分批量查PRODUCT_INV。注意别用超大的IN列表，Cassandra默认限制IN的元素数是1000，超过会报错，所以拆分小批次：

// 第一步：从PRODUCT_BY_STORE拿到目标门店的所有产品编码
Statement getProductsStmt = new SimpleStatement("SELECT Product_Code FROM PRODUCT_BY_STORE WHERE Storecode = ? AND StoreType = ?")
        .setFetchSize(1000);
ResultSet productRs = session.execute(getProductsStmt.bind("601", "Retail"));
Set<String> productCodes = productRs.all().stream()
        .map(row -> row.getString("Product_Code"))
        .collect(Collectors.toSet());

// 第二步：拆分产品编码为小批次，每个批次不超过900个（留点余量）
List<List<String>> codeBatches = Lists.partition(new ArrayList<>(productCodes), 900);
for (List<String> batch : codeBatches) {
    // 构造分区键的元组列表，匹配PRODUCT_INV的主键结构
    List<TupleValue> primaryKeys = batch.stream()
            .map(code -> TupleValue.of(TypeCodec.varchar(), code, "601", "Retail"))
            .collect(Collectors.toList());
    // 用预编译语句提升查询效率
    PreparedStatement invStmt = session.prepare("SELECT * FROM PRODUCT_INV WHERE (Product_Code, Storecode, StoreType) IN ?");
    BoundStatement boundStmt = invStmt.bind(primaryKeys);
    ResultSet invRs = session.execute(boundStmt);
    // 处理结果，比如丢到队列或者直接提交线程池
}

三、多线程处理的正确打开方式

1. 别自己瞎建线程，用线程池！

IO密集型场景（比如Cassandra查询+业务处理），线程数设置为CPU核心数*2或者*4比较合适，避免线程过多导致上下文切换开销。用JDK的ExecutorService就行：

int coreThreads = Runtime.getRuntime().availableProcessors() * 2;
ExecutorService executor = Executors.newFixedThreadPool(coreThreads);

2. 生产者-消费者模式解耦拉取和处理

用线程安全的队列把数据拉取和处理分开，避免拉取速度过快撑爆内存，也能平衡两者的速度：

// 设置队列容量，防止内存溢出
BlockingQueue<Row> dataQueue = new LinkedBlockingQueue<>(1000);

// 生产者线程：负责从Cassandra拉数据入队
executor.submit(() -> {
    try {
        ResultSet rs = session.execute(yourQuery);
        for (Row row : rs) {
            dataQueue.put(row); // 队列满了会自动阻塞，不用自己写逻辑
        }
        // 拉取完毕，放个null标记告诉消费者结束
        dataQueue.put(null);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});

// 消费者线程：多个线程并行处理数据
for (int i = 0; i < coreThreads; i++) {
    executor.submit(() -> {
        try {
            Row row;
            while ((row = dataQueue.take()) != null) {
                // 这里写你的业务处理逻辑，比如解析ProductInfo、更新库存等
                processInventoryRecord(row);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    });
}

// 最后别忘了关闭线程池
executor.shutdown();
executor.awaitTermination(1, TimeUnit.HOURS);

3. 异步查询+回调，榨干IO等待时间

Cassandra Java驱动支持异步查询executeAsync()，返回CompletableFuture，可以在等待查询结果的时间里去做其他事，比同步查询效率更高：

List<CompletableFuture<ResultSet>> queryFutures = new ArrayList<>();
for (List<String> codeBatch : codeBatches) {
    // 构造查询参数
    List<TupleValue> primaryKeys = codeBatch.stream()
            .map(code -> TupleValue.of(TypeCodec.varchar(), code, "601", "Retail"))
            .collect(Collectors.toList());
    PreparedStatement invStmt = session.prepare("SELECT * FROM PRODUCT_INV WHERE (Product_Code, Storecode, StoreType) IN ?");
    // 异步提交查询
    CompletableFuture<ResultSet> future = session.executeAsync(invStmt.bind(primaryKeys));
    queryFutures.add(future);
}

// 等待所有查询完成，然后提交到线程池处理结果
CompletableFuture.allOf(queryFutures.toArray(new CompletableFuture[0])).thenRun(() -> {
    for (CompletableFuture<ResultSet> future : queryFutures) {
        try {
            ResultSet rs = future.get();
            for (Row row : rs) {
                executor.submit(() -> processInventoryRecord(row));
            }
        } catch (Exception e) {
            // 别忘了处理异常，比如记录日志
            log.error("处理库存数据失败", e);
        }
    }
}).join();