Как проверить, действителен ли подписчик, чтобы принять сообщение, полученное по опубликованной теме на MQTT

Question

Я новичок в MQTT.

Я внедряю MQTT в Java, и я использую приведенный ниже код для издателя для публикации в определенной теме,

    public void publish()
        {
            MqttClient myClient = null;
            MqttConnectOptions connOpt;
            try {
                //  Subscription with Brokers
                connOpt = new MqttConnectOptions();
                connOpt.setAutomaticReconnect(true);
                connOpt.setCleanSession(true);//if your not setting cleanSession to false then subscriptions shouldn't be persisted.

                String clientID = UUID.randomUUID().toString().replace("-", "");
                System.out.println("clientID " + clientID);

                myClient = new MqttClient("tcp://192.168.10.500:1883", clientID);
                myClient.connect(connOpt);
                String myTopic = "Device1";
                MqttTopic topic = myClient.getTopic(myTopic);

                int pubQoS = 0;
                MqttMessage message = new MqttMessage("mqttMessage".getBytes());
                message.setQos(pubQoS);
                message.setRetained(false);

                MqttDeliveryToken token = null;
                token = topic.publish(message);
                System.out.println("publish successful with the message :: " + message);
                // Wait until the message has been delivered to the broker
                token.waitForCompletion();
            } catch (MqttException me) {
            } catch (Exception e) {
            }
        }

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

public void subscribe()
{
    try {
        MqttConnectOptions connOpt;

        //  Subscription with mqttBrokerEndPoint
        connOpt = new MqttConnectOptions();
        connOpt.setAutomaticReconnect(true);
        connOpt.setCleanSession(true);//if your not setting cleanSession to false then subscriptions shouldn't be persisted.

        String clientID = UUID.randomUUID().toString().replace("-", "");
        System.out.println("clientID " + clientID);

        MqttSubscriber mqttConnection = new MqttSubscriber();
        myClient = new MqttClient("tcp://192.168.10.500:1883" clientID);
        myClient.setCallback(mqttConnection);
        myClient.connect(connOpt);
        myClient.subscribe("Device1");

    } catch (MqttException e) {
    }
}

@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
    try {
        System.out.println(message);
        boolean isValidClient = true;// Here i need to check if this is the valid subscriber for the message publised on topic "Device1"

        //if(isValidClient) {

            if(message != null) {
                System.out.println("message" + message.toString());
            }

            myClient.unsubscribe("Device1");
            myClient.disconnect();
    //}
  }
    catch(Exception e){}
}

Приведенная выше реализация работает нормально, как есть.

Поскольку я очень плохо знаком с mqtt, у меня естьнекоторые сомнения с вышеупомянутой реализацией.

1) Должен ли идентификатор клиента как в издателе, так и в подписчике быть ОДНЫМ для одного конкретного потока?

или в издателе и подписчике должен отличаться, как указано выше: какой может быть сгенерирован случайным образом?

    String clientID = UUID.randomUUID().toString().replace("-", "");

Этот случайно сгенерированный clientID отлично работает как с подпиской, так и с публикацией.

Но, если я использую один и тот же клиент как для издателя, так и для подписчика и проверяю подписчика?

Я имею в виду использовать"clientID" myClient = new MqttClient(mqttBrokerEndPoint, "clientID"); в подписчике, а затем тот же "clientID" myClient = new MqttClient(mqttBrokerEndPoint, "clientID"); в издателе

Я получаю ошибку сокета ниже в консоли брокера mqtt (используется версия для Windows),

    1549414715: Socket error on client 82, disconnecting.
    1549414715: New client connected from 192.168.10.500 as clientID (c1, k60).
    1549414715: No will message specified.
    1549414715: Sending CONNACK to 82 (0, 0)
    1549414716: New connection from 192.168.10.500 on port 1883.
    1549414716: Client 82 already connected, closing old connection.
    1549414716: Socket error on client 82, disconnecting.
    1549414716: New client connected from 192.168.10.500 as clientID (c1, k60).
    1549414716: No will message specified.
    1549414716: Sending CONNACK to 82 (0, 0)
    1549414716: New connection from 192.168.10.500 on port 1883.
    1549414716: Client 82 already connected, closing old connection.
    1549414716: Socket error on client 82, disconnecting.
    1549414716: New client connected from 192.168.10.500 as clientID (c1, k60).
    1549414716: No will message specified.
    1549414716: Sending CONNACK to 82 (0, 0)

и вышеуказанная программа не работает.

Разве мы не можем использовать один и тот же clientID как для подписчика, так и для издателя?Почему это приводит к ошибке сокета и программам не работает?

2) Обязательно ли имя пользователя и пароль при реализации?Или мы можем установить соединение без указанных ниже двух свойств?

            connOpt.setUserName(USERNAME);
            connOpt.setPassword(PASSWORD.toCharArray());

3) Обязательно ли использование pubQoS для издателя?В настоящее время я использую его как ноль '0'?

        MqttMessage message = new MqttMessage(mqttMessage.getBytes());
        message.setQos(0);
        message.setRetained(false);

Также является ли сохраненный атрибут обязательным для издателя?

4) Являются ли эти 2 атрибута ниже обязательными при подписке?Я использую, как показано ниже в коде.

        connOpt.setAutomaticReconnect(true);
        connOpt.setCleanSession(true);//if your not setting cleanSession to 

false, тогда подписки не должны сохраняться.

5) Кроме того, как только сообщение получено от издателя MQTT для обратного вызова MessageArrived, как показано ниже,как проверить, является ли это действительный подписчик, и продолжить дальнейшую логику?

@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
    try {
        System.out.println(message);
        boolean isValidClient = true;// Here i need to check if this is the valid subscriber for the message publised on topic "Device1"
        //if valid subscriber only i need to read message publised on the topic "Device1"
        **//if(isValidClient) {**

            if(message != null) {
                System.out.println("message" + message.toString());
            }

            myClient.unsubscribe("Device1");
            myClient.disconnect();
    //}
  }
    catch(Exception e){}

Я имею в виду, какой атрибут MQTT API проверяет, что это сообщение отправлено только для этого подписчика, и он может продолжать использовать полученное сообщение.в сообщении поступил обратный вызов?

Т.е. какой атрибут API MQTT можно использовать для проверки того, относится ли полученное сообщение к текущему ПРОЦЕССУ / ШАГУ. (В программе для абонентов, как показано ниже)

ISтема - единственные общие атрибуты между подписчиком и издателем для проверки подписчика в обратном вызове messageArrived?или у нас есть какой-либо другой общий атрибут для проверки действующего договора между подписчиком и издателем?

Или мы должны использовать clientID для проверки подписчика?Но если я использую один и тот же clientID для подписчика и издателя, я получаю ошибку сокета, о которой я упоминал в пункте номер 1.

Как продолжить это?

Answer 1

1) Вы не можете использовать один и тот же идентификатор клиента для двух клиентов mqtt.Идентификаторы клиента для mqtt должны быть уникальными.Вы можете создать другую тему для обработки различных операций.Вы можете сослаться на ответ ниже

Два клиента paho.mqtt, подписывающихся на одного и того же клиента localy

2) Имя пользователя и пароль не являются обязательными.Если ваш сервер mqtt настроен на использование имени пользователя и пароля, тогда вашему клиенту также необходимо передать имя пользователя и пароль на сервер mqtt.Обратитесь к ссылке ниже для понимания использования кода

Как я могу подключить Java-клиент mqtt с именем пользователя и паролем к брокеру emqttd (EMQ)?

Если вытребуется расширенный уровень для обеспечения безопасности tls / ssl

https://dzone.com/articles/secure-communication-with-tls-and-the-mosquitto-broker

3) QOS - уровень качества обслуживания (QoS) - это соглашение между отправителем сообщения и получателемсообщение, определяющее гарантию доставки для конкретного сообщения.Обратитесь к ссылке ниже для получения дополнительной информации о qos

https://www.hivemq.com/blog/mqtt-essentials-part-6-mqtt-quality-of-service-levels/

4) Auto Reconnect (True) - если сервер mqtt отключен во время выполнения, клиент пытается повторно подключиться к серверу.

Чистый сеанс (True) - при повторном подключении к серверу mqtt все более старые сеансы, связанные с тем же идентификатором клиента, будут удалены.

Чистый сеанс (False) - при повторном подключении к серверу mqtt все более старыесеансы, связанные с одним и тем же идентификатором клиента, будут сохранены.Сервер Mqtt в некоторых случаях сохраняет сообщения в зависимости от уровня QOS.

5) Попробуйте создать несколько тем для нескольких операций.В методе прибытия сообщения вы можете проверить, по какой теме пришло сообщение.Затем вы можете вызывать различные методы, основанные на этом.Ниже коды являются одним из способов.Вы можете найти лучший способ, который соответствует вашим потребностям.Cheers !!!!!

@Override
public void deliveryComplete(IMqttDeliveryToken token) 
{
    try 
    {
        if(token.getTopics()[0].equals("TOPIC_A"))
        {
            //Do something
        }   
        else if(token.getTopics()[0].equals("TOPIC_B"))
        {
            //Do something
        }
    } 
    catch (Exception e) 
    {
        MQTTLogger.logSevere(e);
    }
}

Json Формат данных

Первое сообщение

{
  "client-name":"client1"
  "data":""
}

Второе сообщение

{
  "client-name":"client2"
  "data":""
}