我試圖使用基本的Eclipse Paho MQTT客戶端1.1.0版連接到CloudAMQP RabbitMQ實例,訂閱主題並接收消息(我通過網絡發送消息管理控制檯)。Paho Android客戶端在顯示消息時斷開連接
如果應用程序將所有消息載荷發送到日誌輸出,它會很好地工作。
如果應用程序將消息添加到TextView中,則會顯示該消息,但該連接會立即丟棄,並且不會收到更多消息。
完整的項目可在GitHub。下面是一個簡單的例子。
有一個基於服務的MQTT Paho客戶端,但我認爲對於非常簡單的應用程序,基本客戶端應該能夠在Android應用程序UI中接收和顯示消息。
...
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
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 MainActivity extends AppCompatActivity implements MqttCallback {
private static final String TAG = "main";
private Connection connection;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Toolbar toolbar = (Toolbar) findViewById(R.id.toolbar);
setSupportActionBar(toolbar);
configureUI();
}
private Button buttonConnect;
private TextView messageWindow;
private void configureUI() {
buttonConnect = (Button) findViewById(R.id.buttonConnect);
messageWindow = (TextView) findViewById(R.id.messageWindow);
buttonConnect.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
String s = "***";
String d = "test";
String u = "***";
String p = "***";
if (connection != null && connection.isConnected()) {
connection.disconnect();
connection = null;
messageWindow.setText(String.format("Disconnected from server %s",
new Object[]{s}));
return;
}
messageWindow.setText(String.format("Connecting to server %s as user %s",
new Object[]{s, u}));
connection = new Connection(MainActivity.this, MainActivity.this, s, u, p);
connection.connect();
if (connection.isConnected()) {
messageWindow.append("\n\n");
messageWindow.append(String.format("Connected, listening for messages from topic %s",
new Object[]{d}));
connection.subscribe(d);
}
}
});
}
@Override
public void connectionLost(Throwable cause) {
Log.e(TAG, "connectionLost" + cause.getMessage());
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
String msg = new String(message.getPayload());
Log.i(TAG, "Message Arrived: " + msg);
// messageWindow.append(msg);
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
Log.i(TAG, "Delivery Complete!");
}
class Connection {
private static final String TAG = "conn";
private static final String protocol = "tcp://";
private static final int port = 1883;
private static final int version = MqttConnectOptions.MQTT_VERSION_3_1_1;
private static final int keepAliveSeconds = 20 * 60;
private final Context context;
private MqttClient client;
private final String server;
private final String user;
private final String pass;
private final MqttConnectOptions options = new MqttConnectOptions();
public Connection(Context ctx, MqttCallback mqttCallback, String server, String user, String pass) {
this.context = ctx;
this.server = server;
this.user = user;
this.pass = pass;
MqttClientPersistence memPer = new MemoryPersistence();
try {
String url = protocol + server + ":" + port;
client = new MqttClient(url, MqttClient.generateClientId(), memPer);
client.setCallback(mqttCallback);
} catch (MqttException e) {
e.printStackTrace();
}
options.setUserName(user + ":" + user);
options.setPassword(pass.toCharArray());
options.setMqttVersion(version);
options.setKeepAliveInterval(keepAliveSeconds);
}
void connect() {
Log.i(TAG, "buttonConnect");
try {
client.connect(options);
} catch (MqttException ex) {
Log.e(TAG, "Connection attempt failed with reason code = " + ex.getReasonCode() + ":" + ex.getCause());
}
}
public boolean isConnected() {
return client.isConnected();
}
public void disconnect() {
try {
client.disconnect();
} catch (MqttException e) {
Log.e(TAG, "Disconnect failed with reason code = " + e.getReasonCode());
}
}
void subscribe(String dest) {
try {
client.subscribe(dest);
} catch (MqttException e) {
Log.e(TAG, "Subscribe failed with reason code = " + e.getReasonCode());
}
}
}
}
你的意思是「連接被立即被顯示的消息後,收到」「連接被立即丟棄......」 – hardillb
@hardillb我固定它,非常感謝注意到! – mjn