Как сжимать запрос на запись в cassandra при работе с "executeAsync"?

Ответ 1

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

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

Ошибка возврата запроса - это противодавление от Cassandra. Но вы можете выбрать или реализовать 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);
  ....
}

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

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

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

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

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

Ответ 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();
}