我正在学习 Akka Streams，作为练习，我想将原木插入 Cassandra 中。问题是我无法设法使流将日志插入数据库。

我天真地尝试了以下方法:

object Application extends AkkaApp with LogApacheDao {

  // The log file is read line by line
  val source: Source[String, Unit] = Source.fromIterator(() => scala.io.Source.fromFile(filename).getLines())

  // Each line is converted to an ApacheLog object
  val flow: Flow[String, ApacheLog, Unit] = Flow[String]
    .map(rawLine => {
      rawLine.split(",") // implicit conversion Array[String] -> ApacheLog
    })

  // Log objects are inserted to Cassandra
  val sink: Sink[ApacheLog, Future[Unit]] = Sink.foreach[ApacheLog] { log => saveLog(log) }

  source.via(flow).to(sink).run()

}

saveLog（）在LogApacheDao中的定义如下（为了更清晰的代码，我省略了列值）：

val session = cluster.connect()

session.execute(s"CREATE KEYSPACE IF NOT EXISTS $keyspace WITH replication = {'class':'SimpleStrategy', 'replication_factor':1};")

session.execute(s"DROP TABLE IF EXISTS $keyspace.$table;")

session.execute(s"CREATE TABLE $keyspace.$table (...)")

val preparedStatement = session.prepare(s"INSERT INTO $keyspace.$table (...) VALUES (...);")

def saveLog(logEntry: ApacheLog) = {
    val stmt = preparedStatement.bind(...)

    session.executeAsync(stmt)
  }

当进入接收器时，从Array[String]到ApacheLog的转换没有问题（用println验证）。此外，键空间和表都被创建了，但是当执行到SaveLog时，似乎有什么东西被阻塞了，没有插入。

我没有得到任何错误，但Cassandra驱动核心（3.0.0）不断给我：

Connection[/172.17.0.2:9042-1, inFlight=0, closed=false] was inactive for 30 seconds, sending heartbeat
Connection[/172.17.0.2:9042-2, inFlight=0, closed=false] heartbeat query succeeded

我应该补充一点，我用的是dockerized Cassandra。