如何使用“executeAsync”时限制对cassandra的写入请求？

multithreading java backpressure datastax-java-driver rate-limiting

2022-09-03 02:30:20

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

  private final ExecutorService executorService = Executors.newFixedThreadPool(10);

  public void save(String process, int clientid, 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);
        }
      }, executorService);
    } catch (Exception ex) {
      logger.logError("error= ", ex);
    }
  }

我上面的 save 方法将以非常快的速度从多个线程调用。

问题：

我想将请求限制为异步写入Cassandra的方法。如果我的写入速度非常快，超出了我的Cassandra集群可以处理的速度，那么它将开始抛出错误，我希望我的所有写入都应该成功地进入Cassandra而不会有任何损失。executeAsync

我看到这篇文章，解决方案是在固定数量的许可证下使用。但我不确定如何以及什么是实现这一目标的最佳方式。我以前从未使用过信号量。这是逻辑。任何人都可以在我的代码上提供基于信号量的示例，或者是否有任何更好的方法/选项，那么也请告诉我。Semaphore

在编写数据加载程序的上下文中，可以执行如下操作：

为了简单起见，请使用信号量或其他具有固定数量许可的构造（这将是您的最大飞行请求数）。每当您使用 executeAsync 提交查询时，请获取许可证。您实际上只需要 1 个线程（但可能需要引入一个用于执行此操作的 # cpu 内核大小的池），该线程从信号量获取许可并执行查询。它只会阻止获取，直到有可用的许可证。

使用 Futures.addCallback 作为从 executeAsync 返回的未来。回调应该在 onSuccess 和 onFailure 的情况下调用 Sempahore.release（）。通过释放许可证，这应该允许您在步骤 1 中的线程继续并提交下一个请求。

我也看过其他几个帖子，他们谈到了使用，或者哪一个更好，我应该使用？以下是我能想到的选项：RingBufferGuava RateLimitter

使用信号量
使用环形缓冲区
使用番石榴限速器

任何人都可以帮我举个例子，说明我们如何限制请求或为cassandra写入获得背压，并确保所有写入都成功进入cassandra？

答案 1

这不是一个权威的答案，但也许会有所帮助。首先，您应该考虑当无法立即执行查询时会做什么。无论您选择哪种速率限制，如果您收到的请求速率高于写入Cassandra的速率，最终您最终都会被等待的请求堵塞。在那一刻，你需要告诉你的客户将他们的请求保留一段时间（“推回”）。例如，如果它们通过HTTP来，则响应状态将为429“请求太多”。如果在同一进程中生成请求，则确定可接受的最长超时。也就是说，如果Cassandra无法跟上，那么是时候扩展（或调整）它了。

也许在实现速率限制之前，值得在调用方法（使用Thread.sleep（...））之前在线程中进行试验并添加人为延迟，看看它是否解决了您的问题或需要其他东西。save

查询返回错误是来自 Cassandra 的背压。但是，您可以选择或实现重试策略来确定何时重试失败的查询。

此外，还可以查看连接池选项（尤其是监视和调整池）。可以调整每个连接的异步请求数。然而，文档说，对于Cassandra 2.x，这个参数的上限为128，不应该改变它（我会尝试它，尽管:)

使用信号量实现看起来像

/* Share it among all threads or associate with a thread for per-thread limits
   Number of permits is to be tuned depending on acceptable load.
*/
final Semaphore queryPermits = new Semaphore(20); 


public void save(String process, int clientid, long deviceid) {
  ....
  queryPermits.acquire(); // Blocks until a permit is available

  ResultSetFuture future = session.executeAsync(bs);
  Futures.addCallback(future, new FutureCallback<ResultSet>() {
    @Override
    public void onSuccess(ResultSet result) {
      queryPermits.release();
      logger.logInfo("successfully written");
    }
    @Override
    public void onFailure(Throwable t) {
      queryPermits.release(); // Permit should be released in all cases.
      logger.logError("error= ", t);
    }
  }, executorService);
  ....
}

（在实际代码中，我会创建一个装饰器，它将调用包装方法，然后释放许可证。

Guava的RateLimiter类似于信号量，但允许在利用不足期后临时突发，并根据时间（而不是活动查询总数）限制请求。

但是，无论如何，请求都会由于各种原因而失败，因此最好制定一个如何重试它们的计划（如果出现间歇性错误）。

这可能不适合您的情况，但我会尝试使用一些队列或缓冲区来排队请求（例如）。“缓冲区已满”意味着客户端应等待或放弃请求。缓冲区还将用于将失败的请求重新排队。但是，为了更公平，失败的请求可能应该放在队列的前面，以便首先重试它们。此外，还应该以某种方式处理队列已满并同时存在新的失败请求的情况。然后，单线程工作者将从队列中选择请求并将其发送给Cassandra。由于它不应该做太多事情，所以它不太可能成为瓶颈。此工作线程还可以应用自己的速率限制，例如，基于的计时。java.util.concurrent.ArrayBlockingQueuecom.google.common.util.concurrent.RateLimiter

如果有人想尽可能避免丢失消息，他可以在Cassandra面前放置一个具有持久性的消息代理（例如Kafka）。通过这种方式，传入的消息可以在Cassandra的长时间中断中幸存下来。但是，我想，在你的情况下，这是矫枉过正的。

答案 2

只需使用阻塞队列即可。期货是线程化的，回调（成功和失败）将充当使用者，并且无论从哪里调用 save 方法，都将充当生产者。

更好的方法是，您将完整的请求本身放入队列中，并在每次取消排队时逐个将其取消排队。

private final ExecutorService executorService = Executors.newFixedThreadPool(10);

public void save(String process, int clientid, long deviceid, BlockingQueue<Object> queue) {
    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");
          queue.take();
        }

        @Override
        public void onFailure(Throwable t) {
          logger.logError("error= ", t);
          queue.take();
        }
      }, executorService);
    } catch (Exception ex) {
      logger.logError("error= ", ex);
    }
}

public void invokeSaveInLoop(){
    Object dummyObj = new Object();
    BlockingQueue<Object> queue = new ArrayBlockingQueue<>(20);;
    for(int i=0; i< 1000; i++){
        save("process", clientid, deviceid, queue);
        queue.put(dummyObj);
    }
}

如果你想走得更远，在中途检查集群上的负载

public static String getCurrentState(){    
StringBuilder response = new StringBuilder();
            response.append("Current Database Connection Status <br>\n ---------------------------------------------<br>\n");
            final LoadBalancingPolicy loadBalancingPolicy =
                    cluster.getConfiguration().getPolicies().getLoadBalancingPolicy();
            final PoolingOptions poolingOptions =
                    cluster.getConfiguration().getPoolingOptions();
            Session.State state = session.getState();
            for (Host host : state.getConnectedHosts()) {
                HostDistance distance = loadBalancingPolicy.distance(host);
                int connections = state.getOpenConnections(host);
                int inFlightQueries = state.getInFlightQueries(host);
                response.append(String.format("%s current connections=%d, max allowed connections=%d, current load=%d, max load=%d%n",
                                host, connections, poolingOptions.getMaxConnectionsPerHost(distance), inFlightQueries,
                                connections *
                                        poolingOptions.getMaxRequestsPerConnection(distance)))
                        .append("<br>\n");
            }
            return response.toString();
}