我正在为一个项目测试 MQTT。当客户端连接时，我还能够接收有关我的客户端订阅的主题的消息。我已将 QoS 设置为 1，并将 cleanSession 设置为 false。但是当我的客户端再次连接时，我无法接收发送到订阅主题的消息。 在我的应用程序中，几乎所有工作都是由辅助服务完成的。
这是我的代码
AndroidManifest.xml
<?xml version="1.0" encoding="utf-8"?>
<uses-permission android:name="android.permission.WAKE_LOCK" />
<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
<application
android:allowBackup="true"
android:icon="@mipmap/ic_launcher"
android:label="@string/app_name"
android:theme="@style/AppTheme" >
<activity
android:name=".MainActivity"
android:label="@string/app_name"
android:screenOrientation="portrait" >
<intent-filter>
<action android:name="android.intent.action.MAIN" />
<category android:name="android.intent.category.LAUNCHER" />
</intent-filter>
</activity>
<service
android:name=".MqttHelperService"
android:enabled="true"
android:exported="true" />
<!-- MqttService -->
<service android:name="org.eclipse.paho.android.service.MqttService" />
</application>
MainActivity.java
package com.prateek.mqtttest;
import android.app.Activity;
import android.content.Intent;
import android.os.Bundle;
public class MainActivity extends Activity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
startService(new Intent(getBaseContext(), MqttHelperService.class));
}
}
MqttHelperService.java
package com.prateek.mqtttest;
import android.app.Service;
import android.content.Intent;
import android.os.Binder;
import android.os.IBinder;
import android.widget.Toast;
import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
public class MqttHelperService extends Service implements MqttCallback {
private static final String MQTT_URI = "tcp://broker.mqttdashboard.com:1883";
private static final String CLIENT_ID = "prateek";
private static final String MQTT_TOPIC = "mqttmessenger";
private static final int QOS = 1;
private MqttAndroidClient client;
public MqttHelperService() {
}
@Override
public int onStartCommand(Intent intent, int flags, int startId) {
Toast.makeText(this, "MQTT Helper Service Started", Toast.LENGTH_SHORT).show();
new Thread(new Runnable() {
@Override
public void run() {
connect();
}
}, "MqttHelperService").start();
return START_STICKY;
}
public class MqttHelperBinder extends Binder {
public MqttHelperService getService(){
return MqttHelperService.this;
}
}
public void connect() {
client = new MqttAndroidClient(this, MQTT_URI, CLIENT_ID);
client.setCallback(this);
try {
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
client.connect(options, new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken iMqttToken) {
Toast.makeText(getBaseContext(), "connected to MQTT broker", Toast.LENGTH_SHORT).show();
subscribe();
}
@Override
public void onFailure(IMqttToken iMqttToken, Throwable throwable) {
Toast.makeText(getBaseContext(), "failed to connect: " + throwable.getMessage(), Toast.LENGTH_SHORT).show();
}
});
} catch (MqttException e) {
Toast.makeText(this, "could not connect to MQTT broker at " + MQTT_URI, Toast.LENGTH_SHORT).show();
}
}
public void subscribe() {
try {
IMqttToken token = client.subscribe(MQTT_TOPIC, QOS, null, new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken iMqttToken) {
Toast.makeText(getBaseContext(), "subscription successful", Toast.LENGTH_SHORT).show();
}
@Override
public void onFailure(IMqttToken iMqttToken, Throwable throwable) {
Toast.makeText(getBaseContext(), "subscription failed: " + throwable, Toast.LENGTH_SHORT).show();
}
});
} catch (MqttException e) {
Toast.makeText(this, "could not subscribe", Toast.LENGTH_SHORT).show();
}
}
@Override
public IBinder onBind(Intent intent) {
// TODO: Return the communication channel to the service.
throw new UnsupportedOperationException("Not yet implemented");
}
@Override
public void connectionLost(Throwable throwable) {
Toast.makeText(this, "connection lost", Toast.LENGTH_SHORT).show();
}
@Override
public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
Toast.makeText(this, "message received on topic " + s, Toast.LENGTH_SHORT).show();
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
}
@Override
public void onDestroy() {
super.onDestroy();
Toast.makeText(this, "Service Destroyed", Toast.LENGTH_SHORT).show();
}
}
我什至检查了此链接将清除会话标志设置为FALSE，我丢失了已发布的值但在我的代码中找不到错误
我最近也遇到了同样的问题。 现在我认为解决方案非常简单，但我花了很多很多时间才弄清楚。
这行“不好”：
client.connect(mqttOptions, mqqtActionListener);
“正确”是：
client.connect(mqttOptions, null, mqqtActionListener);
如果您使用 2 个参数调用 connect 方法，则您正在使用此构造函数：
public IMqttToken connect(Object userContext, IMqttActionListener callback) throws MqttException
而不是正确的：
public IMqttToken connect(MqttConnectOptions options, Object userContext, IMqttActionListener callback) throws MqttException
我希望这也是你的问题。
最有可能的问题是发布的消息的 QoS=0。订阅和发布都必须是 QoS>0 才能排队等候持久客户端。
您已确认消息是以 QoS 1 发布的，因此接下来要做的就是使用已知的工作工具进行测试，以消除代码中可能存在的问题。我试过：
mosquitto_sub -i prateek -t mqttmessenger -h broker.mqttdashboard.com -v -d -c -q 1
并验证我可以在连接时接收消息：
mosquitto_pub -q 1 -t mqttmessenger -m hello2 -h broker.mqttdashboard.com
然后我断开了 mosquitto_sub 客户端的连接，并使用相同的 mosquitto_pub 命令发布了另一条消息。使用相同的命令重新连接 mosquitto_sub 不会产生任何消息，正如您所看到的。重复该过程，但使用
test.mosquitto.orgbroker.mqttdashboard.com请参阅我的答案 随着清除会话标志设置为 FALSE，我缺少已发布的值。我也遇到过同样的问题并解决了。
您必须在调用 connect 之前调用 mqttClient.setCallback() ，这专门解决了这个细微差别