将数据从文件导入具有 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();
}

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

查询 /x.x.x.1:9042 时出错：com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.1] 池忙（没有可用连接，队列已达到最大大小 256） 查询 /x.x.x.2:9042 时出错：com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.2] 池忙（没有可用连接，队列已达到最大大小 256） 查询 /x.x.x.3:9042 时出错：com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.3] 池忙（没有可用连接，队列已达到其最大大小 256） 查询 /x.x.x.4:9042 时出错：com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.4] 池忙（没有可用的连接，队列已达到其最大大小 256） 查询 /x.x.x.5:9042 时出错：com.datastax.driver.core.exceptions.BusyPoolException: [/x.x.x.5] 池忙（没有可用的连接，队列已达到其最大大小 256）

Answer 1

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

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