java.io.EOFException с пахо

Question

Я хочу сделать стресс-тест на комарах, поэтому я создаю некоторый код, как показано ниже

for (int i = 0; i < 800; i++) {
        final int j = i;
        Thread t = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(j + " : ************");
                try {
                    MqttClient client = new MqttClient("tcp://192.168.88.203", SERVER_CLIENTID_PREFIX + j); 
                    client.connect();

                    MqttMessage message = new MqttMessage((j + ":me").getBytes());
                    message.setQos(2);

                    client.publish(TOPIC_PREFIX + j, message);
                } catch (MqttSecurityException e) {
                    e.printStackTrace();
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        });
        t.start();
    }

Но я получил некоторые ошибки, такие как EOFException во время запуска и какой-то клиент отключен. Я хочу знать, сколько клиентов могут публиковать сообщения одновременно на одном сервере от Mosquitto, и как я могу сделать стресс-тест. Спасибо!

Подробное исключение:

    Connection lost (32109) - java.io.EOFException
    at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:162)
    at java.lang.Thread.run(Thread.java:662)
Caused by: java.io.EOFException
    at java.io.DataInputStream.readByte(DataInputStream.java:250)
    at org.eclipse.paho.client.mqttv3.internal.wire.MqttInputStream.readMqttWireMessage(MqttInputStream.java:51)
    at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:121)
    ... 1 more

И я нашел некоторые журналы с сервера комаров:

1383736170: Socket read error on client Server-82, disconnecting.

Пожалуйста, помогите мне, спасибо!

Answer 1

Я получил эту же ошибку, используя код, аналогичный приведенному выше. Я обнаружил, что изменение QOS на 0 решило проблему.

message.setQos(0);

[Править] Немного больше копания, и я обнаружил, что плагин MQTT для RabbitMQ не поддерживает QOS 2. http://www.rabbitmq.com/mqtt.html

Answer 2

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

Answer 3

Идентификатор клиента является проблемой, генерируя случайный тест

MqttClient.generateClientId ();

Answer 4

В моем случае это было потому, что я случайно использовал tcp://... URL вместо ssl://... и сервер был настроен на запрет небезопасных подключений.

Я также должен был сделать, как сказал @Aidan, и снизить QoS с 2 до 1.

Изменить: я не уверен на 100%, но я думаю, что сервер, который я использую, является RabbitMQ, и это присваивает нестандартное значение значения QoS. Это, наверное, более разумный смысл, если честно:

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

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

Answer 5

Решение - добавить MqttClient.generateClientId

MemoryPersistence persistence = new MemoryPersistence()
MqttClient client = new MqttClient("tcp://192.168.88.203",MqttClient.generateClientId(),persistence); 
client.connect();

или же

MqttClient client = new MqttClient("tcp://192.168.88.203", MqttClient.generateClientId+SERVER_CLIENTID_PREFIX)

Идентификатор должен быть случайным. У меня была эта проблема в скале, и это было мое решение.

Answer 6

Иногда это случается, когда вы пытаетесь отправить большой набор данных. Попробуйте уменьшить размер набора данных. Это решило проблему в моем случае.

Answer 7

В Linux есть ограничение в 1024 файла / сокета, но вы можете его расстроить,ulimit -n 4096 см.: предел подключения MQTT Mosquitto Linux