Как публиковать сообщения MQTT одновременно с помощью Java-клиента? - PullRequest
Новая
       67

Как публиковать сообщения MQTT одновременно с помощью Java-клиента?

0 голосов
/ 08 мая 2019

я пытаюсь публиковать MQTT-сообщения одновременно, используя 5-клиент java, так что каждый java-клиент публикует 1000 сообщений по определенной теме одновременно с брокером MQTT (HIVEMQ)

Я открыл несколько потоков, каждый из которых создает клиент mqtt и подключается к брокеру с помощью ssl и пытается одновременно опубликовать 1000 сообщений, сообщения отправляются, но все подключения не выполняются успешно, и я продолжаю получать исключение 

Client is not connected (32104)
at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:31)
    at org.eclipse.paho.client.mqttv3.internal.ClientComms.sendNoWait(ClientComms.java:199)
    at org.eclipse.paho.client.mqttv3.MqttAsyncClient.publish(MqttAsyncClient.java:1355)
    at org.eclipse.paho.client.mqttv3.MqttClient.publish(MqttClient.java:583)
    at org.eclipse.paho.client.mqttv3.MqttClient.publish(MqttClient.java:575)
    at com.test.MqttPublishSample.publishMessages(MqttPublishSample.java:122)
    at com.test.MqttPublishSample.lambda$start$0(MqttPublishSample.java:74)
    at java.base/java.lang.Thread.run(Thread.java:834)

public class MqttPublishSample {

    public static void main(String... args) throws InterruptedException {

        new MqttPublishSample().start();

    }

  public void start() throws InterruptedException {


        for(int i=0;i<5;i++){

            new Thread(()->{
                MqttClient client = null;
                try {
                    client = obtainConnection();//code to obtain connection using MqttClient
                    publishMessages(client);//code to publish message using simple for loop 

                } catch (MqttException e) {
                    e.printStackTrace();
                }

            }).start();
        }
    }

public MqttClient obtainConnection() throws MqttException {
        String clientId = "sslTestClient"+ThreadLocalRandom.current().nextInt(0,5);
        MqttClient client = null;
        try {
            client = new MqttClient("ssl://localhost:8883", clientId, new MemoryPersistence());
        } catch (MqttException e) {
            e.printStackTrace();
        }

        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName("user1");
        mqttConnectOptions.setPassword("pass1".toCharArray());
        try {
            mqttConnectOptions.setSocketFactory(getTruststoreFactory());
        } catch (Exception e) {
            e.printStackTrace();
        }

        System.out.println("connecting...");
        client.connect(mqttConnectOptions);
        return client;
    }

Я ожидаю, что все клиенты успешно подключатся к брокеру и опубликуют сообщение без исключения

1 Ответ

1 голос
/ 08 мая 2019

Возможно, вы используете один и тот же clientID в вашем потоке, поэтому сервер отключит дубликат. Поскольку вы используете LocalThreadRandom, существует вероятность столкновения (достаточно большого, так как есть только 5 вариантов). Вы могли бы использовать уникальный идентификатор, предоставленный generateClientId (), или совместно использовать метод между потоками, которые отслеживают их.

...