Снижение производительности Vert.x при запуске с опцией -cluster

Question

Мне интересно, если кто-то испытал ту же проблему.

У нас есть приложение Vert.x, и в конце его цель - вставить 600 миллионов строк в кластер Cassandra. Мы тестируем скорость Vert.x в сочетании с Cassandra, проводя тесты в меньших количествах.

Если мы запустим жирную банку (с помощью плагина Shade) без опции -cluster, мы сможем вставить 10 миллионов записей за минуту. Когда мы добавляем параметр -cluster (в конце концов мы будем запускать приложение Vert.x в кластере), для вставки 10 миллионов записей требуется около 5 минут.

Кто-нибудь знает, почему?

Мы знаем, что конфигурация Hazelcast создаст некоторые накладные расходы, но никогда не думала, что это будет в 5 раз медленнее. Это означает, что нам понадобится 5 экземпляров EC2 в кластере, чтобы получить тот же результат при использовании 1 EC2 без опции кластера.

Как уже упоминалось, все работает на EC2 случаях:

• 2 Cassandra серверы t2.small
сервер
• 1 Vert.x на t2.2xlarge

Answer 1

Вы на самом деле работаете в угловых случаях менеджера кластеров Vert.x Hazelcast.

Прежде всего, вы используете рабочий Verticle для отправки сообщений (30000001). Под капотом Hazelcast блокируется, и, таким образом, когда вы отправляете сообщение от работника, версия 3.3.3 не принимает это в учетной записи. Недавно мы добавили это исправление https://github.com/vert-x3/issues/issues/75 (нет в 3.4.0.Beta1, но присутствует в 3.4.0-SNAPSHOTS), что улучшит этот случай.

Во-вторых, когда вы отправляете все свои сообщения одновременно, оно запускается в другой угловой регистр, который предотвращает использование кластерным менеджером Hazelcast кеша топологии кластера.Этот кэш топологии обычно обновляется после того, как первое сообщение было отправлено и отправка всех сообщений одним выстрелом предотвращает использование ache (краткое описание HazelcastAsyncMultiMap # getInProgressCount будет> 0 и предотвратит использование кеша), следовательно, уплата штрафа дорогостоящего поиска (отсюда и кеш).

Если я использую репродуктор Bertjan с 3.4.0-SNAPSHOT + Hazelcast и следующее изменение: отправьте сообщение в пункт назначения, дождитесь ответа. После ответа отправьте все сообщения, после чего я получаю много улучшений.

Без кластеризация: 5852 мс С кластеризация с ГЦ 3.3.3: 16745 мс С кластеризация с HZ 3.4.0-SNAPSHOT + начальное сообщение: 8609 мс

Я считаю, что и вы не должны использовать работник Verticle чтобы послать это много сообщений и вместо этого отправить их с помощью цикла цикла событий через партии. Возможно, вам следует объяснить ваш прецедент, и мы можем подумать о том, как его решить.

Answer 2

Просто добавить код проект. Думаю, это поможет.

Отправитель Verticle:

public class ProviderVerticle extends AbstractVerticle { 

    @Override 
    public void start() throws Exception { 
     IntStream.range(1, 30000001).parallel().forEach(i -> { 
     vertx.eventBus().send("clustertest1", Json.encode(new TestCluster1(i, "abc", LocalDateTime.now()))); 
     }); 
    } 

    @Override 
    public void stop() throws Exception { 
     super.stop(); 
    } 
} 

И вставка Verticle

public class ReceiverVerticle extends AbstractVerticle { 

    private int messagesReceived = 1; 

    private Session cassandraSession; 

    @Override 
    public void start() throws Exception { 

     PoolingOptions poolingOptions = new PoolingOptions() 
       .setCoreConnectionsPerHost(HostDistance.LOCAL, 2) 
       .setMaxConnectionsPerHost(HostDistance.LOCAL, 3) 
       .setCoreConnectionsPerHost(HostDistance.REMOTE, 1) 
       .setMaxConnectionsPerHost(HostDistance.REMOTE, 3) 
       .setMaxRequestsPerConnection(HostDistance.LOCAL, 20) 
       .setMaxQueueSize(32768) 
       .setMaxRequestsPerConnection(HostDistance.REMOTE, 20); 

     Cluster cluster = Cluster.builder() 
       .withPoolingOptions(poolingOptions) 
       .addContactPoints(ClusterSetup.SEEDS) 
       .build(); 

     System.out.println("Connecting session"); 
     cassandraSession = cluster.connect("kiespees"); 
     System.out.println("Session connected:\n\tcluster [" + cassandraSession.getCluster().getClusterName() + "]"); 
     System.out.println("Connected hosts: "); 

     cassandraSession.getState().getConnectedHosts().forEach(host -> System.out.println(host.getAddress())); 

     PreparedStatement prepared = cassandraSession.prepare(
       "insert into clustertest1 (id, value, created) " + 
         "values (:id, :value, :created)"); 

     PreparedStatement preparedTimer = cassandraSession.prepare(
       "insert into timer (name, created_on, amount) " + 
         "values (:name, :createdOn, :amount)"); 

     BoundStatement timerStart = preparedTimer.bind() 
       .setString("name", "clusterteststart") 
       .setInt("amount", 0) 
       .setTimestamp("createdOn", new Timestamp(new Date().getTime())); 
     cassandraSession.executeAsync(timerStart); 

     EventBus bus = vertx.eventBus(); 

     System.out.println("Bus info: " + bus.toString()); 
     MessageConsumer<String> cons = bus.consumer("clustertest1"); 
     System.out.println("Consumer info: " + cons.address()); 

     System.out.println("Waiting for messages"); 

     cons.handler(message -> { 
      TestCluster1 tc = Json.decodeValue(message.body(), TestCluster1.class); 

      if (messagesReceived % 100000 == 0) 
       System.out.println("Message received: " + messagesReceived); 

      BoundStatement boundRecord = prepared.bind() 
        .setInt("id", tc.getId()) 
        .setString("value", tc.getValue()) 
        .setTimestamp("created", new Timestamp(new Date().getTime())); 
      cassandraSession.executeAsync(boundRecord); 

      if (messagesReceived % 100000 == 0) { 
       BoundStatement timerStop = preparedTimer.bind() 
         .setString("name", "clusterteststop") 
         .setInt("amount", messagesReceived) 
         .setTimestamp("createdOn", new Timestamp(new Date().getTime())); 
       cassandraSession.executeAsync(timerStop); 
      } 

      messagesReceived++; 
      //message.reply("OK"); 
     }); 
    } 

    @Override 
    public void stop() throws Exception { 
     super.stop(); 
     cassandraSession.close(); 
    } 
} 

Answer 3

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

Например ваш текущий поток (без кластеризации) что-то вроде:

client -> 
    vert.x app -> 
    in memory same process eventbus (negletible) -> 
    handler -> cassandra 
    <- vert.x app 
<- client 

После включения кластеризации:

client -> 
    vert.x app -> 
    serialize request -> 
     network request cluster member -> 
     deserialize request -> 
      handler -> cassandra 
     <- serialize response 
     <- network reply 
    <- deserialize response 
    <- vert.x app 
<- client 

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

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