如何从Cassandra流式读取全量记录，超150万条出现read timeout如何解决？

• 你设置的fetchSize为100000过大，单次请求需要拉取10万行数据，Cassandra服务端无法在默认超时时间内完成数据加载和返回，触发超时。
• 直接用原生SimpleStatement执行全表SELECT DISTINCT查询，没有做流控和背压处理，大数据量下请求压力直接打向Cassandra节点，容易引发节点响应超时。
• 驱动默认的读超时配置不足以支撑百万级以上的全表扫描请求。

1. 降低fetchSize到合理区间 #

将fetchSize调整为1000~5000，避免单次请求拉取过多数据，同时可以单独给语句设置更长的超时时间：

val selectDistinctPersistenceIds = new SimpleStatement(
  "SELECT DISTINCT persistence_id, partition_nr FROM messages")
  .setFetchSize(2000)
  .setReadTimeoutMillis(60000) // 单独给该语句设置1分钟读超时

2. 使用官方内置的持久化ID查询流（优先推荐） #

akka-persistence-cassandra已经内置了优化过的持久化ID查询流，自带分页、背压、重试逻辑，不需要自己手写查询语句：

val querier = PersistenceQuery(system)
  .readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)

// currentPersistenceIds会拉取全量持久化ID后结束流，persistenceIds是持续监听新增ID的流
val persistenceIdsSource: Source[String, NotUsed] = querier.currentPersistenceIds()

3. 调整Cassandra驱动超时配置 #

在application.conf中修改Cassandra连接的超时配置，适配大数据量查询场景：

cassandra-journal {
  socket {
    read-timeout = 60s
    connect-timeout = 30s
  }
}

cassandra-snapshot-store {
  socket {
    read-timeout = 60s
    connect-timeout = 30s
  }
}

4. 增加重试机制 #

如果偶发超时仍存在，可以给流增加指数退避重试逻辑：

import akka.stream.contrib.Retry
import scala.concurrent.duration._

val retriedSource = Retry(
  maxRetries = 3,
  minBackoff = 1.second,
  maxBackoff = 10.seconds,
  randomFactor = 0.2
)(() => querier.currentPersistenceIds())

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