java - 将数据从文件导入到具有 5 个节点的 Cassandra 集群会导致 BusyConnectionException

Question

对于我的论文，我需要将文件中的数据上传到 Cassandra 集群。使用 session.execute() 太慢了。所以我决定使用session.executeAsyn()。但它会导致 BusyConnectionException。

这是我的 Java 代码:

    final PoolingOptions poolingOptions = new PoolingOptions();
    poolingOptions.setMaxRequestsPerConnection(HostDistance.LOCAL, 32768)
            .setMaxRequestsPerConnection(HostDistance.REMOTE, 32768);
    final Cluster cluster = Cluster.builder()
            .withPoolingOptions(poolingOptions)
            .addContactPoint("x.x.x.x")
            .withPort(9042)
            .build();
    final Session session = cluster.connect();
    System.out.println("session object---" + session.getState());
    final String path = "&PathToFile%";
    final File dir = new File(path);

    session.execute("use products;");
    for (final File file : dir.listFiles()) {
        final BufferedReader br = new BufferedReader(new FileReader(file));
        String str;
        final String insert = br.readLine();
        while ((str = br.readLine()) != null) {
            final String query = insert + str.substring(0, str.length() - 1) + "IF NOT EXISTS ;";
            session.executeAsync(query);
        }
    }
    session.close();
    cluster.close();
}

以下是我执行代码时遇到的异常:

Error querying /x.x.x.1:9042 : com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.1] Pool is busy (no available connection and the queue has reached its max size 256) Error querying /x.x.x.2:9042 : com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.2] Pool is busy (no available connection and the queue has reached its max size 256) Error querying /x.x.x.3:9042 : com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.3] Pool is busy (no available connection and the queue has reached its max size 256) Error querying /x.x.x.4:9042 : com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.4] Pool is busy (no available connection and the queue has reached its max size 256) Error querying /x.x.x.5:9042 : com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.5] Pool is busy (no available connection and the queue has reached its max size 256)

Answer 1

当您在一个连接上发出过多请求时，会发生繁忙异常。您需要控制发送的请求数量。最简单的方法是使用信号量或类似的东西。我have a class它包装了 Session 并允许控制正在进行的请求数量，因此它的行为类似于异步，直到达到限制，并且会阻塞，直到正在进行的请求数量低于限制。您可以使用我的代码，或实现类似的东西。

更新:您正在使用 light-weight transactions (LWT) (IF NOT EXISTS 子句)，这会严重影响集群的性能，因为每个插入都需要与其他节点协调...