Как дросселировать запрос записи в кассандру при работе с executeAsync?

Не авторитетный ответ, но, возможно, он был бы полезен. Сначала вы должны подумать, что бы вы сделали, если запрос не может быть выполнен сразу. Независимо от того, какое ограничение скорости вы выбрали, если вы получаете запросы с большей скоростью, чем вы можете писать в Cassandra, в конечном итоге вы забьете свой процесс ожидающими запросами. И в этот момент вам нужно будет сказать вашим клиентам, чтобы они на время задерживали свои запросы («оттолкнуть»). Например. если они поступают через HTTP, статус ответа будет 429 «Слишком много запросов». Если вы генерируете запросы в одном процессе, тогда решите, какой максимальный тайм-аут будет приемлемым. Тем не менее, если Кассандра не успевает, то пора ее масштабировать (или настраивать).

Возможно, перед введением ограничений скорости стоит поэкспериментировать и добавить искусственные задержки в ваши потоки перед вызовом метода save (с использованием Thread.sleep (...)) и посмотреть, решит ли он вашу проблему или нужно что-то еще.

Ошибка возврата запроса является противодавлением от Кассандры. Но вы можете выбрать или реализовать RetryPolicy, чтобы определить, когда повторять неудачные запросы.

Также вы можете посмотреть параметры пула соединений (и особенно Мониторинг и настройка пула). Можно настроить количество асинхронных запросов на соединение. Однако в документации говорится, что для 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);
  ....
}

(В реальном коде я бы создал обратный вызов оболочки, который бы выпускал разрешения, а затем вызывал обернутые методы)

RateLimiter Guava похож на семафор, но допускает временные всплески после периодов недостаточного использования и ограничивает запросы на основе времени (а не общего количества активных запросов).

Однако запросы все равно не будут выполняться по разным причинам, поэтому, вероятно, лучше иметь план, как повторить их (в случае периодических ошибок).

В вашем случае это может быть неприемлемо, но я бы попытался использовать какую-нибудь очередь или буфер для постановки запросов в очередь (например, java.util.concurrent.ArrayBlockingQueue). «Буфер заполнен» означает, что клиенты должны подождать или отказаться от запроса. Буфер также будет использоваться для повторной постановки в очередь неудавшихся запросов. Однако, чтобы быть более справедливым, неудавшиеся запросы, вероятно, следует помещать в начало очереди, чтобы они сначала выполнялись повторно. Также нужно как-то справляться с ситуацией, когда очередь заполнена и одновременно появляются новые неудавшиеся запросы. Затем однопоточный воркер будет выбирать запросы из очереди и отправлять их в Cassandra. Поскольку это не должно делать много, маловероятно, что это станет узким местом. Этот воркер также может применять свои собственные ограничения скорости, например на основе времени с com.google.common.util.concurrent.RateLimiter.

Если кто-то хочет максимально избежать потери сообщений, он может поставить посредника сообщений с настойчивостью (например, Kafka) перед Cassandra. Таким образом, входящие сообщения могут выдержать даже длительные перерывы в работе Cassandra. Но, думаю, в твоем случае это излишество.

Простое использование очереди с блокировкой должно подойти. Фьючерсы являются многопоточными, и обратный вызов (успешный и неудачный) будет действовать как потребитель, и где бы вы ни вызывали метод сохранения, он будет действовать как производитель.

Еще лучше будет, если вы поместите полный запрос в очередь и удаляете его один за другим, сохраняя при каждом удалении.

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();
}