我正在为 android 构建一个简单的 MQTT 客户端，并且在 RMBS 控制台上出现“客户端标识符的套接字错误”错误。这只发生在客户端的 android 实现中（我还创建了一个 Java 桌面客户端并且运行没有问题）。对于 Android 客户端，我正在起诉 Paho Java 客户端库。这是我的代码：
这是安卓客户端：
package com.example.mqttdroid;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.internal.MemoryPersistence;
import android.app.Activity;
import android.os.AsyncTask;
import android.os.Bundle;
import android.util.Log;
import android.view.Menu;
import android.widget.Toast;
public class MQTTClient extends Activity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_mqttclient);
// new BackTask().execute(); not used because (seems to be the problem)
MqttConnectOptions conOpts = new MqttConnectOptions();
conOpts.setKeepAliveInterval(30);
conOpts.setWill(client.getTopic("Error"), "something bad happened".getBytes(), 1, true);
client.connect(conOpts);
client.subscribe("/House/Kitchen/Bulb");
client.setCallback( new MqttCallback() {
@Override
public void connectionLost(Throwable arg0) {
// TODO Auto-generated method stub
}
@Override
public void deliveryComplete(IMqttDeliveryToken arg0) {
// TODO Auto-generated method stub
}
@Override
public void messageArrived(String arg0, MqttMessage arg1)
throws Exception {
// TODO Auto-generated method stub
Toast.makeText(Main.this, arg0.toString(), Toast.LENGTH_SHORT).show();
}
});
}
@Override
public boolean onCreateOptionsMenu(Menu menu) {
// Inflate the menu; this adds items to the action bar if it is present.
getMenuInflater().inflate(R.menu.mqttclient, menu);
return true;
}
/*public class BackTask extends AsyncTask<Void, Void, Void>{
private MqttClient client;
@Override
protected Void doInBackground(Void... params) {
// TODO Auto-generated method stub
try {
client = new MqttClient("tcp://"Ip of machine running RSMB":1883", "ANDROID1", new MemoryPersistence());
client.connect();
client.subscribe("House/Kitchen/Bulb");
} catch (MqttException e) {
// TODO Auto-generated catch block
Log.e("ERROR", "NOT CONNECTED");
e.printStackTrace();
}
return null;
}
@Override
protected void onPostExecute(Void result) {
// TODO Auto-generated method stub
super.onPostExecute(result);
try {
client.setCallback( new MqttCallback() {
@Override
public void messageArrived(MqttTopic arg0, MqttMessage arg1)
throws Exception {
// TODO Auto-generated method stub
Toast.makeText(MQTTClient.this, arg0.toString(), Toast.LENGTH_SHORT).show();
}
@Override
public void deliveryComplete(MqttDeliveryToken arg0) {
// TODO Auto-generated method stub
}
@Override
public void connectionLost(Throwable arg0) {
// TODO Auto-generated method stub
}
});
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}*/
这是桌面 Java 客户端：
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.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.internal.MemoryPersistence;
public class MQTTBaseClass {
/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
MqttClientPersistence persistence;
try {
MqttClient client = new MqttClient("tcp://localhost:1883", "PC",new MemoryPersistence());
MqttConnectOptions conOpts = new MqttConnectOptions();
conOpts.setKeepAliveInterval(30);
conOpts.setWill(client.getTopic("Error"), "something bad happened".getBytes(), 1, true);
client.connect(conOpts);
MqttMessage msg = new MqttMessage("hello".getBytes());
msg.setQos(0);
msg.setRetained(true);
MqttTopic topic = client.getTopic("House/Kitchen/Bulb");
client.subscribe("House/Kitchen/Bulb");
try {
client.setCallback( new MqttCallback() {
@Override
public void messageArrived(MqttTopic arg0, MqttMessage arg1)
throws Exception {
// TODO Auto-generated method stub
System.out.println(arg1.toString());
}
@Override
public void deliveryComplete(MqttDeliveryToken arg0) {
// TODO Auto-generated method stub
}
@Override
public void connectionLost(Throwable arg0) {
// TODO Auto-generated method stub
}
});
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
topic.publish(msg);
} catch (MqttPersistenceException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
只是一些注意事项：
我通过 WiFi 连接到我的 android 设备，当我运行 Java Dekstop 客户端时，我的桌面也是如此。
Java Destop 客户端与 RSMB 在同一台机器上运行
Java 桌面客户端创建并订阅主题“House/Kitchen/Bulb”并发送带有字符串“Hello”的消息
Android 客户端还订阅“House/Kitchen/Bulb”并尝试显示带有接收到的消息的 Toast。
我在 android manifest 上添加了 Internet 权限
android 设备似乎可以很好地连接到代理，但是一旦我初始化 Java 桌面服务客户端（或 Eclipse 中的 Paho 客户端插件并发布消息），我就会收到提到的错误。
我在 RSMB 运行的同一台机器上使用模拟器运行了应用程序，我得到了同样的错误。
可能是什么问题？
更新：
最初，我遇到了“主线程上的网络”异常，因此我将连接操作移至 AsyncTask。现在，当我使用 Java 客户端发布消息时，Android 客户端似乎仍处于连接状态（Asynctask 可能一直在创建问题），但似乎没有调用 MqttCallback() 的 messageArrived()。
更新 2：
我设法使它工作。这是我现在使用的代码：
package com.example.mqttphone;
*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.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import android.app.Activity;
import android.os.AsyncTask;
import android.os.Bundle;
import android.util.Log;
import android.view.Menu;
import android.widget.Toast;*
public class Main extends Activity {
protected static String msg;
public MqttClient client;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
try {
client = new MqttClient("tcp://10.1.201.27:1883", "ANDROID1", new MemoryPersistence());
MqttConnectOptions conOpts = new MqttConnectOptions();
conOpts.setKeepAliveInterval(30);
conOpts.setWill(client.getTopic("Error"), "something bad happened".getBytes(), 1, true);
client.setCallback( new MqttCallback() {
@Override
public void connectionLost(Throwable arg0) {
// TODO Auto-generated method stub
}
@Override
public void deliveryComplete(IMqttDeliveryToken arg0) {
// TODO Auto-generated method stub
}
@Override
public void messageArrived(String arg0, MqttMessage arg1)
throws Exception {
// TODO Auto-generated method stub
Main.msg = arg1.toString();
Main.this.runOnUiThread(new Runnable() {
@Override
public void run() {
// TODO Auto-generated method stub
Toast.makeText(Main.this, msg, Toast.LENGTH_LONG).show();
}
});
Log.e("MESSAGE RECEIVED", arg1.toString());
}
});
client.connect(conOpts);
//MqttMessage msg = new MqttMessage("ANDROID MESSAGE".getBytes());
//client.getTopic("world").publish(msg);
if(client.isConnected()){
client.subscribe("/House/Kitchen/Bulb");
Toast.makeText(this, "CONNECTED", Toast.LENGTH_SHORT).show();
}
} catch (MqttException e) {
// TODO Auto-generated catch block
Log.e("ERROR", "NOT CONNECTED");
e.printStackTrace();
}
}
@Override
public boolean onCreateOptionsMenu(Menu menu) {
// Inflate the menu; this adds items to the action bar if it is present.
getMenuInflater().inflate(R.menu.main, menu);
return true;
}
}