使用Cassandra的executeAsync时出现数据丢失问题求助

你这段代码出现数据丢失的核心原因是没有正确处理executeAsync的异步结果，且并发控制不当，具体问题点如下：

• 未等待异步操作完成：executeAsync返回的ResultSetFuture代表异步数据库操作的结果，但你的代码只是调用了这个方法，没有等待它完成，也没有处理可能的失败。外层的Future{...}会立即完成，根本没等到数据库插入操作结束，导致部分请求还没执行完就被丢弃。
• 无限制并发请求：直接对100万条数据用Future.sequence发起全量并发请求，远远超过了Cassandra驱动的连接池上限（默认连接数一般为几十到几百），大量请求会被驱动拒绝或者超时，最终导致插入失败。
• 冗余的Future包装：writeToCassandra里用Future{...}包裹executeAsync完全多余，executeAsync本身已经返回异步结果，额外包装会掩盖操作的真实状态。
• SQL字符串拼接风险：直接用字符串拼接生成SQL，不仅有SQL注入隐患，还可能因为特殊字符导致插入失败（虽然这里status是固定的on/off，但也是不良实践）。

修正后的代码示例 #

import scala.concurrent.{Future, ExecutionContext}
import com.datastax.oss.driver.api.core.CqlSession
import com.datastax.oss.driver.api.core.cql.PreparedStatement

case class Sensor(id: Int, status: String)

def writeToCassandra(sensor: Sensor, session: CqlSession, preparedStmt: PreparedStatement)(implicit ec: ExecutionContext): Future[Unit] = {
  // 绑定参数到预编译语句
  val boundStmt = preparedStmt.bind(sensor.id, sensor.status)
  // 将DataStax异步结果转为Scala Future，等待插入操作完成
  Future.fromTry(session.executeAsync(boundStmt).toScalaFuture.map(_ => ()))
}

def batchInsert(session: CqlSession)(implicit ec: ExecutionContext): Future[Unit] = {
  // 预编译SQL，提升性能并避免注入风险
  val preparedStmt = session.prepare("insert into alpakka.sensor (id, status) values(?, ?)")
  
  val dataList = 1 to 1000000
  // 控制并发数，根据集群和驱动配置调整（示例为100）
  val parallelism = 100
  
  // 分批次处理请求，避免并发过载
  Future.traverse(dataList.grouped(parallelism)) { batch =>
    Future.sequence(batch.map { row =>
      val status = if (Math.round(Math.random() * 10) % 2 == 1) "on" else "off"
      val sensor = Sensor(id = row, status = status)
      writeToCassandra(sensor, session, preparedStmt)
        // 捕获插入异常，便于排查失败请求
        .recover { case e: Exception =>
          println(s"插入失败，id: ${sensor.id}，错误信息: ${e.getMessage}")
        }
    })
  }.map(_ => ())
}

关键优化点说明 #

• 预编译语句：使用PreparedStatement绑定参数，既避免SQL注入，又能让Cassandra缓存查询计划，大幅提升插入性能。
• 正确处理异步结果：通过toScalaFuture将DataStax的异步结果转为Scala标准Future，确保插入动作真正执行完毕后才标记完成。
• 并发控制：通过grouped将数据分成批次，每次并行处理固定数量的请求，避免超出驱动连接池和Cassandra集群的处理能力。
• 错误处理：添加recover捕获异常，可快速定位失败的插入请求，也可扩展实现重试逻辑（比如基于重试次数或间隔的自定义重试机制）。

内容的提问来源于stack exchange，提问作者김정완