了解您需要调用 client.setCallback 并实现 MqttCallbackHandler 以便接收有关您订阅的主题的消息非常重要。
这是一个可以发布和订阅等的代码示例。
以下代码最初发布 mqtt 主题和有效负载：
- 主题：安卓手机
- Payload：您好，我是Android Mqtt Client。
该代码订阅主题“测试者”。如果它接收到主题为“tester”且有效负载为“警报激活”的消息，则它会发布以下主题和有效负载（通过上述回调）：
- 主题： Fitlet
- 有效负载：您好，Mosquitto Broker 收到您的消息说警报已激活。
如果您使用的是 Mosquitto，那么终端中的以下命令将导致此消息被触发：
mosquitto_pub -h 192.168.9.100 -t tester -m "Alarm Activated" -u fred -P 1234
我的 Mosquitto 用户名是 fred，密码是 1234
编码：
package colin.android.mqtt;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
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.MqttCallbackExtended;
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.MqttPersistenceException;
import java.io.UnsupportedEncodingException;
public class MainActivity extends AppCompatActivity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
String clientId = MqttClient.generateClientId();
//The URL of the Mosquitto Broker is 192.168.9.100:1883
final MqttAndroidClient client = new MqttAndroidClient(this.getApplicationContext(), "tcp://192.168.9.100:1883", clientId);
client.setCallback(new MqttCallbackHandler(client));//This is here for when a message is received
MqttConnectOptions options = new MqttConnectOptions();
try {
options.setUserName("fred");
options.setPassword("1234".toCharArray());
IMqttToken token = client.connect(options);
token.setActionCallback(new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken asyncActionToken) {
// We are connected
Log.d("mqtt", "onSuccess");
//-----------------------------------------------------------------------------------------------
//PUBLISH THE MESSAGE
MqttMessage message = new MqttMessage("Hello, I am an Android Mqtt Client.".getBytes());
message.setQos(2);
message.setRetained(false);
String topic = "AndroidPhone";
try {
client.publish(topic, message);
Log.i("mqtt", "Message published");
// client.disconnect();
//Log.i("mqtt", "client disconnected");
} catch (MqttPersistenceException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
//-----------------------------------------------------------------------------------------------
String subtopic = "tester";
int qos = 1;
try {
IMqttToken subToken = client.subscribe(subtopic, qos);
subToken.setActionCallback(new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken asyncActionToken) {
// The message was published
Log.i("mqtt", "subscription success");
}
@Override
public void onFailure(IMqttToken asyncActionToken,
Throwable exception) {
// The subscription could not be performed, maybe the user was not
// authorized to subscribe on the specified topic e.g. using wildcards
Log.i("mqtt", "subscription failed");
}
});
} catch (MqttException e) {
e.printStackTrace();
}
//---------------------------------------------------------------------------
}
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
// Something went wrong e.g. connection timeout or firewall problems
Log.d("mqtt", "onFailure");
}
});
} catch (MqttException e) {
e.printStackTrace();
}
}
}//End of Activity class
//-----------------------------------------------------------------------------
class MqttCallbackHandler implements MqttCallbackExtended {
private final MqttAndroidClient client;
public MqttCallbackHandler (MqttAndroidClient client)
{
this.client=client;
}
@Override
public void connectComplete(boolean b, String s) {
Log.w("mqtt", s);
}
@Override
public void connectionLost(Throwable throwable) {
}
public void AlarmActivatedMessageReceived()
{
MqttMessage msg= new MqttMessage("Hello, the Mosquitto Broker got your message saying that the Alarm is Activated.".getBytes());
try {
this.client.publish("Fitlet", msg);
Log.i("mqtt", "Message published");
} catch (MqttPersistenceException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
Log.w("mqtt", mqttMessage.toString());
if (mqttMessage.toString().contains("Alarm Activated"))
{
AlarmActivatedMessageReceived();
}
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
}
}