java - 斯卡拉- Cassandra : cluster read fails with error "Can' t use this Cluster instance because it was previously closed"

Question

使用 datastax 驱动程序从 5 节点集群中的表读取时出现此错误。

2015-02-19 03:24:09,908 ERROR [akka.actor.default-dispatcher-9] OneForOneStrategy akka://user/HealthServiceChecker-49e686b9-e189-48e3-9aeb-a574c875a8ab Can't use this Cluster instance because it was previously closed java.lang.IllegalStateException: Can't use this Cluster instance because it was previously closed at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1128) ~[cassandra-driver-core-2.0.4.jar:na] at com.datastax.driver.core.Cluster.init(Cluster.java:149) ~[cassandra-driver-core-2.0.4.jar:na] at com.datastax.driver.core.Cluster.connect(Cluster.java:225) ~[cassandra-driver-core-2.0.4.jar:na] at com.datastax.driver.core.Cluster.connect(Cluster.java:258) ~[cassandra-driver-core-2.0.4.jar:na]

我能够使用 cqlsh 进行连接并执行读取操作。

任何线索可能是这里的问题？

设置:

• 一致性级别:ONE

• 键空间复制策略:'类':'NetworkTopologyStrategy'，'DC2':'1'，'DC1': '1'

• Cassandra 版本:2.0.6

管理 Cassandra session 的代码是核心，它也是；

trait ConfigCassandraCluster
  extends CassandraCluster
{
  def cassandraConf: CassandraConfig
  lazy val port = cassandraConf.port
  lazy val host = cassandraConf.host
  lazy val cluster: Cluster =
    Cluster.builder()
      .addContactPoints(host)
      .withReconnectionPolicy(new ExponentialReconnectionPolicy(100, 30000))
      .withPort(port)
      .withSocketOptions(new SocketOptions().setKeepAlive(true))
      .build()

  lazy val keyspace = cassandraConf.keyspace
  private lazy val casSession = cluster.connect(keyspace)
  val session = new SessionProvider(casSession)
}

class SessionProvider(casSession: => Session) extends Logging {
  var lastSuccessful: Long = 0
  var firstSuccessful: Long = -1
  def apply[T](fn: Session => T): T = {
    val result = retry(fn, 15)
    if(firstSuccessful < 0)
      firstSuccessful = System.currentTimeMillis()
    lastSuccessful = System.currentTimeMillis()
    result
  }

  private def retry[T](fn: Session => T, remainingAttempts: Int): T = {
    //retry logic
}

Answer 1

问题是，如果遇到 NoHostAvailableException，cluster.connect(keyspace) 将自行关闭集群。由于在重试逻辑期间，您会遇到 IllegalStateException。

看看 Cluster init() 方法，您会了解更多。

您的问题的解决方案是，在重试逻辑中，执行 Cluster.builder.addContactPoint(node).build.connect(keyspace)。这将在您重试时启用一个新的集群对象。