使用datastax java驱动程序异步写入cassandra的有效方法？

Question

我正在使用datastax java驱动程序3.1.0连接到cassandra集群，我的cassandra集群版本是2.0.10。我正在异步地写，具有定额一致性。

  public void save(final String process, final int clientid, final long deviceid) {
    String sql = "insert into storage (process, clientid, deviceid) values (?, ?, ?)";
    try {
      BoundStatement bs = CacheStatement.getInstance().getStatement(sql);
      bs.setConsistencyLevel(ConsistencyLevel.QUORUM);
      bs.setString(0, process);
      bs.setInt(1, clientid);
      bs.setLong(2, deviceid);

      ResultSetFuture future = session.executeAsync(bs);
      Futures.addCallback(future, new FutureCallback<ResultSet>() {
        @Override
        public void onSuccess(ResultSet result) {
          logger.logInfo("successfully written");
        }

        @Override
        public void onFailure(Throwable t) {
          logger.logError("error= ", t);
        }
      }, Executors.newFixedThreadPool(10));
    } catch (Exception ex) {
      logger.logError("error= ", ex);
    }
  }

下面是我的缓存语句类：

public class CacheStatement {
  private static final Map<String, PreparedStatement> cache =
      new ConcurrentHashMap<>();

  private static class Holder {
    private static final CacheStatement INSTANCE = new CacheStatement();
  }

  public static CacheStatement getInstance() {
    return Holder.INSTANCE;
  }

  private CacheStatement() {}

  public BoundStatement getStatement(String cql) {
    Session session = CassUtils.getInstance().getSession();
    PreparedStatement ps = cache.get(cql);
    // no statement cached, create one and cache it now.
    if (ps == null) {
      synchronized (this) {
        ps = cache.get(cql);
        if (ps == null) {
          cache.put(cql, session.prepare(cql));
        }
      }
    }
    return ps.bind();
  }
}

我上面的保存方法将从多个线程调用，我认为Bound语句不是线程安全的。BtwStatementCache类是线程安全的，如上所示。

• 由于Bound语句不是线程安全的。如果我从多个线程异步编写，我上面的代码会有问题吗？
• 其次，我在addCallback参数中使用了Executors.newFixedThreadPool（10）。这样可以吗？还是会有任何问题？或者我应该使用MoreExecutors.directExecutor。那么这两者之间有什么区别？最好的方法是什么？

下面是我使用datastax java驱动程序连接到cassandra的连接设置：

Builder builder = Cluster.builder();
    cluster =
        builder
            .addContactPoints(servers.toArray(new String[servers.size()]))
            .withRetryPolicy(new LoggingRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE))
            .withPoolingOptions(poolingOptions)
            .withReconnectionPolicy(new ConstantReconnectionPolicy(100L))
            .withLoadBalancingPolicy(
                DCAwareRoundRobinPolicy
                    .builder()
                    .withLocalDc(
                        !TestUtils.isProd() ? "DC2" : TestUtils.getCurrentLocation()
                            .get().name().toLowerCase()).withUsedHostsPerRemoteDc(3).build())
            .withCredentials(username, password).build();

Answer 1

我认为你正在做的很好。您可以通过在应用程序启动时准备所有语句来进一步优化，这样您就可以缓存所有内容，这样您就不会在“保存”时因准备语句而受到任何性能影响，并且您不会锁定工作流中的任何内容。

绑定语句不是线程安全的，但准备语句是的，并且每次调用 get 语句时，您都会返回一个新的绑定语句。实际上，预准备语句的 .bind（） 函数实际上是新绑定语句 （ps）.bind（） 的快捷方式。并且您没有从多个线程访问相同的绑定语句。所以你的代码很好。

相反，对于线程池，您实际上是在每个addCallback函数上创建一个新的线程池。这是浪费资源。我不使用这种回调方法，我更喜欢自己管理纯FutureResultSet，但我在datastax留档上看到了使用MoreExecutors.sameThreadExecutor（）而不是MoreExecutors.directExecutor（）的示例。