メッセージMQTTがサブスクライブ・クライアントで受信しない

Question

MQTTプロトコルで作業しています.2つの異なるJavaアプリケーションを使用してパブリッシュおよびサブスクライブを試みています。

私の最初のアプリケーションは「公開」です。私はメッセージをMQTTサーバー上に公開しています。

私の2番目のアプリケーションは「購読」です。トピックを購読してメッセージを受け取ろうとしています。しかし、私はメッセージを受け取ることはありません。

私は2つのアプリケーションを実行するとき、私は "サブスクライブ"アプリケーションと "公開"アプリケーションを実行した後に開始します。 「公開」アプリケーションが開始されると、「購読」アプリケーションへの接続が失われ、メッセージを受信できなくなります。

"Subscribe"アプリケーションでは、私のメソッドmessageArrived（）がclient.setCallback（this）によって呼び出されることはありません。 （下のコードを参照してください）。ここで

私の2のコードのアプリケーションである：（公開）メイン

package publishclient; 

import java.util.Random; 
import org.eclipse.paho.client.mqttv3.MqttClient; 
import org.eclipse.paho.client.mqttv3.MqttConnectOptions; 
import org.eclipse.paho.client.mqttv3.MqttCallback; 
import org.eclipse.paho.client.mqttv3.MqttMessage; 
import org.eclipse.paho.client.mqttv3.MqttException; 
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; 
import org.eclipse.paho.client.mqttv3.MqttAsyncClient; 
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; 



public class PubClient implements MqttCallback { 

MqttClient client; 
MqttConnectOptions connOpt; 

Random rand = new Random(); 
int nbRandom = 0; 
int valMax =151, valMin = 40; 

public PubClient() throws MqttException { 
String broker = "tcp://88.177.147.17:1883"; // Adress MQTT Server 
String clientId = "0bdd-4445-82f3-928f8ddb1887"; // ClientID 
String topic1f = "ApplicationRio/capteur"; // Topic 
int QoSserveur = 2; 

     try { 
      String uuid = "ac8da3c6-0bdd-4445-82f3-928f8ddb3294"; 

      MemoryPersistence persistence = new MemoryPersistence(); 

      // Create 2 objects : client and connOpt 
      client = new MqttClient(broker, clientId, persistence); 
      connOpt = new MqttConnectOptions(); 

      connOpt.setCleanSession(true); 

      client.setCallback(this); 

      // Connection to MQTT server 
      System.out.println("Connexion a : " + broker + " Publisher"); 
      client.connect(connOpt); 

      //Create random number for my message 
      nbRandom = valMin + rand.nextInt(valMax-valMin); 
      System.out.println("nb aleatoire = " + nbRandom); 
      String messageAEnvoyer = uuid + "//" + nbRandom; 
      System.out.println("Message a envoyer : " + messageAEnvoyer); 

      MqttMessage message = new MqttMessage(); 
      message.setPayload(messageAEnvoyer.getBytes()); 
      message.setQos(QoSserveur); 
      client.publish(topic1f, message); 

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

    @Override 
    public void connectionLost(Throwable thrwbl) {System.out.println("Perdue connexion");} 

    @Override 
    public void messageArrived(String string, MqttMessage mm) throws Exception { 
    System.out.println("Message recu est : "+ new String(mm.getPayload()));} 

    @Override 
    public void deliveryComplete(IMqttDeliveryToken imdt) { 
    System.out.println("Message delivre au broker"); 
    } 
} 

： クラスPubClient：

アプリケーションの公開

package publishclient; 

import org.eclipse.paho.client.mqttv3.MqttException; 

public class PublishClient { 

public static void main(String[] args) throws MqttException {  
    PubClient publieur = new PubClient(); 
} 

"購読" アプリケーション： クラスサブクライアント：

package subscribeclient; 

import org.eclipse.paho.client.mqttv3.MqttClient; 
import org.eclipse.paho.client.mqttv3.MqttConnectOptions; 
import org.eclipse.paho.client.mqttv3.MqttCallback; 
import org.eclipse.paho.client.mqttv3.MqttMessage; 
import org.eclipse.paho.client.mqttv3.MqttException; 
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; 

public class SubClient implements MqttCallback { 

MqttClient clientsub; 
MqttConnectOptions connOpt; 


public SubClient() throws MqttException{ 
String broker = "tcp://88.177.147.17:1883"; // Adress MQTT Server 
String clientId = "0bdd-4445-82f3-928f8ddb1887"; // ClientID 
String topic1f = "ApplicationRio/capteur"; // Topic 
int QoSserveur = 2; 

try{ 

      // Create 2 objects : client and connOpt 
      clientsub = new MqttClient(broker, clientId); 
      connOpt = new MqttConnectOptions(); 

      connOpt.setCleanSession(false); 
      connOpt.setKeepAliveInterval(30); 

      clientsub.setCallback(this); 

      // Connection to MQTT Server 
      System.out.println("Connexion a : " + broker + " Subscriber"); 
      clientsub.connect(connOpt); 


      clientsub.subscribe(topic1f,QoSserveur); 

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

    @Override 
    public void connectionLost(Throwable thrwbl) { 
     System.out.println("Connexion perdue"); 
    } 

    @Override 
    public void messageArrived(String string, MqttMessage message) throws Exception { 
     System.out.println("Le message recu est : " + new String(message.getPayload())); 
    } 

    @Override 
    public void deliveryComplete(IMqttDeliveryToken imdt) { 
     System.out.println("Message arrive"); 
    } 
} 

メイン（購読）：

package subscribeclient; 

import org.eclipse.paho.client.mqttv3.MqttException; 

public class SubscribeClient { 

public static void main(String[] args) throws MqttException { 
    SubClient subscriber = new SubClient(); 
} 

} 

私の2つのアプリケーションは、同じ時間で実行する必要がある、と私はアプリケーションのすべての時間を実行するので、私は切断する必要はありません。

"Publish Client"を実行したときに私の "Subscribe Client"が切断され、なぜ "Subscribe Message"に自分のメッセージが届かないのか分かりましたか？

私はMQTTのライブラリにorg.eclipse.paho.client.mqttv3-1.0.2.jarを使用します。

Answer 1

クライアントIDは、すべてのクライアント間で一意である必要があります。パブリッシャとサブスクライバに同じクライアントIDを使用しているため、パブリッシャが接続するときにブローカがサブスクライバを解放します。