我在通过 eclipse Paho 客户端在 Moquette 服务器中使用离线 MQTT 消息时遇到问题。
以下是我遵循的步骤。
以上步骤成功,没有任何问题。
然后我停止了我的消费者应用程序,并将 MQTT 数据发送到具有相同主题的代理。使用我的发布者应用程序 - 服务器能够接收这些消息,但此时没有任何消费者使用此消息,因为我已经停止了我的消费者。 然后我再次启动我的消费者应用程序。它已成功连接到代理,但是,它没有收到我在消费者关闭时发送给代理的任何消息。
我是否需要对 Moquette 服务器进行任何特定配置才能保留数据(使用干净会话: false)? 还是我错过了什么?
请在下面找到我的示例代码,
Moquette 服务器初始化
package com.gbids.mqtt.moquette.main;
import com.gbids.mqtt.moquette.server.PublishInterceptor;
import io.moquette.interception.InterceptHandler;
import io.moquette.server.Server;
import io.moquette.server.config.IConfig;
import io.moquette.server.config.MemoryConfig;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
public class ServerLauncher {
public static void main(String[] args) throws IOException {
Properties props = new Properties();
final IConfig configs = new MemoryConfig(props);
final Server mqttBroker = new Server();
final List<? extends InterceptHandler> userHandlers = Arrays.asList(new PublishInterceptor());
mqttBroker.startServer(configs, userHandlers);
System.out.println("moquette mqtt broker started, press ctrl-c to shutdown..");
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
System.out.println("stopping moquette mqtt broker..");
mqttBroker.stopServer();
System.out.println("moquette mqtt broker stopped");
}
});
}
}
MQTT 消费者
package com.gbids.mqtt.moquette.main;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class ConsumerLauncher implements MqttCallback {
private static final String topicPrefix = "devices/reported";
private static final String broker = "tcp://0.0.0.0:1883";
private static final String clientIdPrefix = "consumer";
public static void main(String[] args) throws MqttException {
final String clientId = "consumer_1";
MqttClient sampleClient = new MqttClient(broker, clientId, new MemoryPersistence());
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setCleanSession(false);
sampleClient.connect(connOpts);
sampleClient.subscribe(topicPrefix + "/#", 1);
sampleClient.setCallback(new ConsumerLauncher());
}
public void connectionLost(Throwable throwable) {
System.out.println("Consumer connection lost : " + throwable.getMessage());
}
public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
System.out.println("Message arrived from topic : " + s + " | Message : " + new String(mqttMessage.getPayload()) + " | Message ID : " +mqttMessage.getId());
}
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
System.out.println("Delivery completed from : " + clientIdPrefix + "_1");
}
}
MQTT 发布者
package com.gbids.mqtt.moquette.main;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class ClientLauncher {
private static final String content = "{\"randomData\": 25}";
private static final String willContent = "Client disconnected unexpectedly";
private static final String broker = "tcp://0.0.0.0:1883";
private static final String clientIdPrefix = "client";
public static void main(String[] args) throws Exception{
sendDataWithQOSOne();
System.exit(0);
}
private static void sendDataWithQOSOne(){
try {
final String clientId = "client_1";
MqttClient sampleClient = new MqttClient(broker, clientId, new MemoryPersistence());
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setCleanSession(false); // for publisher - this is not needed I think
sampleClient.connect(connOpts);
MqttMessage message = new MqttMessage(content.getBytes());
message.setQos(1);
final String topic = "devices/reported/" + clientId;
sampleClient.publish(topic, message);
System.out.println("Message published from : " + clientId + " with payload of : " + content);
sampleClient.disconnect();
} catch (MqttException me) {
me.printStackTrace();
}
}
}
在您的情况下,您需要在
retained
(发布者)中创建 true
时将 MqttMessage
标志设置为 ClientLauncher
。默认值为 false
,如 文档中所示。
...
message.setRetained(true)
...
设置此标志可以使消息保留在代理上并发送到新连接的客户端。请注意,代理仅保留主题的最后一条消息。无法为特定主题保留多条消息。
请参阅我的答案 随着清除会话标志设置为 FALSE,我缺少已发布的值。我也遇到过同样的问题并解决了。
您必须在调用 connect 之前调用 mqttClient.setCallback() ,这专门解决了这个细微差别