Масштабирование связей Mqtt с mosquitto broker

Question

Я пытаюсь установить несколько (100k) соединений mqtt с mktt broker, но не может выходить за пределы 5000, используя клиент paho java.

У меня есть хорошие указатели из другого сообщения, но я столкнулся с ошибкой, так как я пересекаю 5000 нитей. Ошибка: java.lang.OutOfMemoryError: не удалось создать новый собственный поток

У меня есть экземпляр Ubuntu RAM/16VCPU 16 гигабайт в облаке, поэтому ресурсы здесь не похожи.

Max MQTT connections

Мой текущий ULIMIT устанавливается ниже.

ulimit -a 
core file size   (blocks, -c) 0 
data seg size   (kbytes, -d) unlimited 
scheduling priority    (-e) 0 
file size    (blocks, -f) unlimited 
pending signals     (-i) 128311 
max locked memory  (kbytes, -l) 64 
max memory size   (kbytes, -m) unlimited 
open files      (-n) 30000000 
pipe size   (512 bytes, -p) 8 
POSIX message queues  (bytes, -q) 819200 
real-time priority    (-r) 0 
stack size    (kbytes, -s) 8192 
cpu time    (seconds, -t) unlimited 
max user processes    (-u) 200000 
virtual memory   (kbytes, -v) unlimited 
file locks      (-x) unlimited 

Кроме того, сообщение рассказывает о добавлении 21 виртуального адреса. На этом облачном экземпляре у меня установлен eth0.cfg:

---------------------- 
# The primary network interface 
auto eth0 
iface eth0 inet dhcp 
--------------------------- 

Как добавить сюда виртуальные IP-адреса?

Любые указатели здесь помогут.

Спасибо.

Редактировать: Вот мой код класса потока - public class SendMessage extends Thread { public String clientId; public String topicId; public String имя пользователя; public String password; public String payloadType;

SendMessage(String topicId,String clientId,String username,String password,String payloadType) { 
     this.topicId = topicId; 
     this.password = password; 
     this.username = username; 
     this.clientId = clientId; 
     this.payloadType = payloadType; 
    } 
    SendMessage(String topicId) { 
     this.topicId = topicId; 
    } 
    public void run() { 


    // Perform the requested action 
     try { 

    ArrayList<MqttClient> sampleClient = new ArrayList<MqttClient>(); 
    MqttConnectOptions connOpts = new MqttConnectOptions(); 
    connOpts.setCleanSession(true); 
    connOpts.setPassword(password.toCharArray()); 
    connOpts.setUserName(username); 

    for (int i = 0; i < connectionCountPerThread ; i++) { 
    MqttClient newClient = new MqttClient("tcp://" + url, clientId + "_" +i); 
       sampleClient.add(newClient); 
       newClient.connect(connOpts); 

      } 


      MqttMessage message1; 
      int qos = 0; 
      MockObservations mo = new MockObservations(); 
      for (int i = 0 ; i < msgCount ; i++) { 
      for (int conn = 0; conn < connectionCountPerThread ; conn++) { 

      message1 = new MqttMessage(mo.getSensorReading(payloadType).getBytes()); 
      message1.setQos(qos); 
      sampleClient.get(conn).publish(topicId, message1); 

      } 

      } 
      for (int conn = 0; conn < connectionCountPerThread ; conn++) { 
      sampleClient.get(conn).disconnect(); 
      } 

     } catch(MqttException me) { 
      System.out.println("reason "+me.getReasonCode()); 
      System.out.println("msg "+me.getMessage()); 
      System.out.println("loc "+me.getLocalizedMessage()); 
      System.out.println("cause "+me.getCause()); 
      System.out.println("excep "+me); 
      me.printStackTrace(); 
     } 
    } 
} 

Answer 1

Вы создали слишком много потоков. Paho использует, насколько я знаю, 2 потока на отдельное соединение. Вы не заходите слишком далеко с моделью «Thread for Connection». Я считаю, что единственный способ создать такой инструмент тестирования нагрузки - это использовать подход NIO с Java.

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

Это может помочь вам: https://en.wikipedia.org/wiki/C10k_problem