时间戳

首页 > 解决方案 > 将数据从文件导入具有 5 个节点的 Cassandra 集群导致 BusyConnectionException

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

问题描述

对于我的论文，我需要将数据从文件上传到 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();
}

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

查询 /xxx1:9042 时出错：com.datastax.driver.core.exceptions.BusyPoolException：[/xxx1] 池正忙（无可用连接，队列已达到其最大大小 256） 查询 /xxx2:9042 时出错：com.datastax .driver.core.exceptions.BusyPoolException：[/xxx2] 池正忙（无可用连接，队列已达到其最大大小 256）查询 /xxx3:9042 时出错：com.datastax.driver.core.exceptions.BusyPoolException：[ /xxx3] 池忙（无可用连接，队列已达到最大大小 256） 查询 /xxx4:9042 时出错：com.datastax.driver.core.exceptions.BusyPoolException：[/xxx4] 池忙（无可用连接并且队列已达到其最大大小 256) 查询 /xxx5:9042 时出错：com.datastax.driver.core.exceptions.BusyPoolException: [/xxx5] 池忙（无可用连接，队列已达到其最大大小 256）

标签： javacassandraconnection-poolingdatastax-java-driver

解决方案

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

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

推荐阅读