为什么使用 ParparedStatement、BoundStatement、Session.executeAsync 和 ResultSetFuture 的应用程序会消耗如此高的 CPU？答案

【问题标题】：why a application using ParparedStatement, BoundStatement, Session.executeAsync and ResultSetFuture consuming such high CPU?为什么使用 ParparedStatement、BoundStatement、Session.executeAsync 和 ResultSetFuture 的应用程序会消耗如此高的 CPU？
【发布时间】：2016-04-03 23:19:27
【问题描述】：

我需要在 3 个节点的 Cassandra(2.1.11) 集群中写入近 1000 万条记录，复制因子为 1 ，我的步骤几乎与datastax的Java驱动程序如下：

    String  insert_query = "insert into " + keyspace + "." + tblName
            + " (a, b, c, d,"
            + "e, f, g, h, i, j,"
            + "k, l, m, n)  VALUES "
            + "(?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
   List<ResultSetFuture> futures = new ArrayList<ResultSetFuture>();
   PreparedStatement statement = session.prepare(insert_query);
   BoundStatement bind = null;
   int max = 5000000 or 6000000 or 7000000 
   for(int i = 0; i < max ; i++) {
                bind = statement.bind(
                        id,
                        ...                 
                        null,
                        null,
                        null
                );
            ResultSetFuture resultSetFuture = session.executeAsync(bind);
            futures.add(resultSetFuture);
  } //for
 for(ResultSetFuture future : futures){
            future.getUninterruptibly(15000, TimeUnit.MILLISECONDS);
  }

然后，我的应用程序在 16 核的机器下运行，我监控进程的 CPU 消耗：

PID   USER     PR  NI  VIRT   RES  SHR S  %CPU    %MEM  TIME+     COMMAND
25502 pengcz   20  0   30.8g  27g  19m S  1263.7  25.8  104:28.82    java

我发现cpu使用率太高(1263.7%)，而且发现cpu使用率高持续时间更长，甚至写多条记录都写失败了。

不知道是不是我的错误步骤导致了这种情况？任何建议将不胜感激！

【问题讨论】：

标签： java cassandra datastax-java-driver

【解决方案1】：

您实际上是一次提交 5、6 或 700 万个请求，然后等待所有请求一次完成。由于您正在构建一个包含请求数量的期货列表，因此您不仅提交了许多请求，而且还使用了大量内存来跟踪这些响应。我想在不等待响应的情况下提交许多请求会产生大量 CPU 来生成请求有效负载并将它们写入网络，此外还有越来越大的 GC 压力使这些未来保持在一个巨大的列表中。

你应该做的是一次只有这么多的飞行请求（也许500个？），等待它们完成，然后提交下一组，依此类推。肯定有比这更好和更优化的技术，而且这也不能很好地处理错误情况，但它应该会降低你的内存占用并防止你用许多请求锤击 C*。下面是一个示例：

    String insert_query = "insert into " + keyspace + "." + tblName
            + " (a, b, c, d,"
            + "e, f, g, h, i, j,"
            + "k, l, m, n)  VALUES "
            + "(?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
    List<ResultSetFuture> futures = new ArrayList<ResultSetFuture>();
    PreparedStatement statement = session.prepare(insert_query);
    BoundStatement bind = null;
    int max = 1000000;

    for (int i = 0; i < max; i++) {
        bind = statement.bind(i,
                null,
                null,
                null
        );
        ResultSetFuture resultSetFuture = session.executeAsync(bind);
        futures.add(resultSetFuture);

        if (futures.size() % 500 == 0 || i == max - 1) {
            for (ResultSetFuture future : futures) {
                future.getUninterruptibly(15000, TimeUnit.MILLISECONDS);
            }
            futures.clear();
        }
    }

如需进一步指导，请查看Asynchronous queries with the Java driver。

【讨论】：