https://blog.csdn.net/hbw020/article/details/79865397
Android mqtt入門(mén) Android studio
2018年04月09日 14:02:30?hbw020?閱讀數(shù)：1564
分享 mqtt簡(jiǎn)單使用介紹：
1、as創(chuàng)建工程
2奕坟、官網(wǎng)下載mqtt支持包放入lib文件悔政，點(diǎn)擊打開(kāi)鏈接，https://repo.eclipse.org/content/repositories/paho-releases/org/eclipse/paho/org.eclipse.paho.client.mqttv3/1.2.0/罩引。當(dāng)前使用的是：
org.eclipse.paho.client.mqttv3-1.2.0.jar
3嘹悼、然后開(kāi)始工作：
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 java.util.concurrent.Executors;import java.util.concurrent.ScheduledExecutorService;import java.util.concurrent.TimeUnit;/** * 全局主服務(wù)分發(fā)和接收mqtt消息 */public class MQTTServiceextends Service {private static final StringTAG ="MQTTService";? ? public final StringSERVICE_CLASSNAME ="de.eclipsemagazin.mqtt.push.MQTTService";? ? private Handlerhandler;? ? //? ? private String host = "tcp://192.168.2.151:1883";//? ? private String userName = "admin";//? ? private String passWord = "password";? ? private int i =1;? ? private static MqttClientclient;? ? //? ? private String myTopic = "qaiot/user/f5c71e03-c324-4b32-823c-563471e86da9";//? ? private String myTopic = "qaiot/user/f5c71e03-c324-4b32";? ? private StringmyTopic ="qaiot/server/user/1.0/cn";? ? StringclientId ="qaiot/user/f5c71e03-c324-4b32";? ? private MqttConnectOptionsoptions;? ? private static ScheduledExecutorServicescheduler;? ? public JooanMQTTService() {? ? }@Override? ? public IBinderonBind(Intent intent) {// TODO: Return the communication channel to the service.? ? ? ? throw new UnsupportedOperationException("Not yet implemented");? ? }@Override? ? public void onCreate() {super.onCreate();? ? ? ? initMqttClient();? ? ? ? handler =new Handler() {@Override? ? ? ? ? ? public void handleMessage(Message msg) {super.handleMessage(msg);? ? ? ? ? ? ? ? if (msg.what ==1) {? ? ? ? ? ? ? ? ? ? Toast.makeText(JooanApplication.get(), (String) msg.obj, Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? ? ? NotificationManager manager = (NotificationManager) getSystemService(Context.NOTIFICATION_SERVICE);? ? ? ? ? ? ? ? ? ? Notification notification =new Notification(R.drawable.menu_item_small_bg, "Mqtt即時(shí)推送", System.currentTimeMillis());//? ? ? ? ? ? ? ? ? ? notification.contentView = new RemoteViews("com.hxht.testmqttclient", R.layout.activity_notification);? ? ? ? ? ? ? ? ? ? notification.contentView =new RemoteViews("com.jooan.qiaoanzhilian", R.layout.service_jooan_mqtt_notification);? ? ? ? ? ? ? ? ? ? notification.contentView.setTextViewText(R.id.tv_desc, (String) msg.obj);? ? ? ? ? ? ? ? ? ? notification.defaults = Notification.DEFAULT_SOUND;? ? ? ? ? ? ? ? ? ? notification.flags = Notification.FLAG_AUTO_CANCEL;? ? ? ? ? ? ? ? ? ? manager.notify(i++, notification);? ? ? ? ? ? ? ? ? ? Log.e(TAG, "Mqtt收到推送結(jié)果: " + (String) msg.obj);? ? ? ? ? ? ? ? }else if (msg.what ==2) {? ? ? ? ? ? ? ? ? ? Log.i(TAG, "連接成功");? ? ? ? ? ? ? ? ? ? Toast.makeText(JooanApplication.get(), "連接成功", Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? ? ? try {client.subscribe(clientId, 1);? ? ? ? ? ? ? ? ? ? }catch (Exception e) {? ? ? ? ? ? ? ? ? ? ? ? e.printStackTrace();? ? ? ? ? ? ? ? ? ? }? ? ? ? ? ? ? ? }else if (msg.what ==3) {? ? ? ? ? ? ? ? ? ? Toast.makeText(JooanApplication.get(), "連接失敗嵌器，系統(tǒng)正在重連", Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? ? ? Log.i(TAG, "連接失敗肛真，系統(tǒng)正在重連");? ? ? ? ? ? ? ? }? ? ? ? ? ? }? ? ? ? };? ? ? ? startReconnect();? ? }private void initMqttClient() {try {//host為主機(jī)名，test為clientid即連接MQTT的客戶端ID爽航，一般以客戶端唯一標(biāo)識(shí)符表示蚓让，MemoryPersistence設(shè)置clientid的保存形式，默認(rèn)為以內(nèi)存保存? ? ? ? ? ? client =new MqttClient(JooanApplication.get().BROKER_URL, myTopic, new MemoryPersistence());? ? ? ? ? ? //MQTT的連接設(shè)置? ? ? ? ? ? options =new MqttConnectOptions();? ? ? ? ? ? //設(shè)置是否清空session,這里如果設(shè)置為false表示服務(wù)器會(huì)保留客戶端的連接記錄讥珍，這里設(shè)置為true表示每次連接到服務(wù)器都以新的身份連接? ? ? ? ? ? options.setCleanSession(true);? ? ? ? ? ? //設(shè)置連接的用戶名? ? ? ? ? ? options.setUserName(JooanApplication.get().userName);? ? ? ? ? ? //設(shè)置連接的密碼? ? ? ? ? ? options.setPassword(JooanApplication.get().passWord.toCharArray());? ? ? ? ? ? // 設(shè)置超時(shí)時(shí)間 單位為秒? ? ? ? ? ? options.setConnectionTimeout(10);? ? ? ? ? ? // 設(shè)置會(huì)話心跳時(shí)間 單位為秒 服務(wù)器會(huì)每隔1.5*20秒的時(shí)間向客戶端發(fā)送個(gè)消息判斷客戶端是否在線凭疮，但這個(gè)方法并沒(méi)有重連的機(jī)制? ? ? ? ? ? options.setKeepAliveInterval(20);//? ? ? ? ? ? options.setWill();//如果項(xiàng)目中需要知道客戶端是否掉線可以調(diào)用該方法? ? ? ? ? ? //設(shè)置回調(diào)? ? ? ? ? ? client.setCallback(new MqttCallback() {@Override? ? ? ? ? ? ? ? public void connectionLost(Throwable cause) {//連接丟失后，一般在這里面進(jìn)行重連? ? ? ? ? ? ? ? ? ? Log.w(TAG, "connectionLost----------: " + cause.getMessage());? ? ? ? ? ? ? ? }@Override? ? ? ? ? ? ? ? public void messageArrived(String topic, MqttMessage message)throws Exception {//subscribe后得到的消息會(huì)執(zhí)行到這里面? ? ? ? ? ? ? ? ? ? Log.w(TAG, "messageArrived---------- ");? ? ? ? ? ? ? ? ? ? Message msg =new Message();? ? ? ? ? ? ? ? ? ? msg.what =1;? ? ? ? ? ? ? ? ? ? msg.obj = topic +"---" + message.toString();? ? ? ? ? ? ? ? ? ? handler.sendMessage(msg);? ? ? ? ? ? ? ? }@Override? ? ? ? ? ? ? ? public void deliveryComplete(IMqttDeliveryToken token) {//publish后會(huì)執(zhí)行到這里? ? ? ? ? ? ? ? ? ? Log.w(TAG, "deliveryComplete---------: " + token.isComplete());? ? ? ? ? ? ? ? }? ? ? ? ? ? });? ? ? ? }catch (Exception e) {? ? ? ? ? ? e.printStackTrace();? ? ? ? }? ? }private void startReconnect() {scheduler = Executors.newSingleThreadScheduledExecutor();? ? ? ? scheduler.scheduleAtFixedRate(new Runnable() {@Override? ? ? ? ? ? public void run() {if (!client.isConnected()) {? ? ? ? ? ? ? ? ? ? connect();? ? ? ? ? ? ? ? }? ? ? ? ? ? }? ? ? ? }, 0 *1000, 10 *1000, TimeUnit.MILLISECONDS);? ? }private void connect() {new Thread(new Runnable() {@Override? ? ? ? ? ? public void run() {try {client.connect(options);? ? ? ? ? ? ? ? ? ? Message msg =new Message();? ? ? ? ? ? ? ? ? ? msg.what =2;? ? ? ? ? ? ? ? ? ? handler.sendMessage(msg);? ? ? ? ? ? ? ? }catch (Exception e) {? ? ? ? ? ? ? ? ? ? e.printStackTrace();? ? ? ? ? ? ? ? ? ? Message msg =new Message();? ? ? ? ? ? ? ? ? ? msg.what =3;? ? ? ? ? ? ? ? ? ? handler.sendMessage(msg);? ? ? ? ? ? ? ? }? ? ? ? ? ? }? ? ? ? }).start();? ? }//? ? 判斷服務(wù)是否運(yùn)行中? ? private boolean serviceIsRunning() {? ? ? ? ActivityManager manager = (ActivityManager) getSystemService(Context.ACTIVITY_SERVICE);? ? ? ? for (ActivityManager.RunningServiceInfo service : manager.getRunningServices(Integer.MAX_VALUE)) {if (SERVICE_CLASSNAME.equals(service.service.getClassName())) {return true;? ? ? ? ? ? }? ? ? ? }return false;? ? }public static void stopJooanMQTTService() {if (client !=null) {try {client.disconnect();? ? ? ? ? ? }catch (Exception e) {? ? ? ? ? ? ? ? e.printStackTrace();? ? ? ? ? ? }? ? ? ? }if (scheduler !=null) {try {scheduler.shutdown();? ? ? ? ? ? }catch (Exception e) {? ? ? ? ? ? ? ? e.printStackTrace();? ? ? ? ? ? }? ? ? ? }? ? }}
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;import org.eclipse.paho.client.mqttv3.MqttCallback;import org.eclipse.paho.client.mqttv3.MqttException;import org.eclipse.paho.client.mqttv3.MqttMessage;/** * 發(fā)布消息的回調(diào)類 * * 必須實(shí)現(xiàn)MqttCallback的接口并實(shí)現(xiàn)對(duì)應(yīng)的相關(guān)接口方法 *? ? ? ?CallBack 類將實(shí)現(xiàn) MqttCallBack串述。每個(gè)客戶機(jī)標(biāo)識(shí)都需要一個(gè)回調(diào)實(shí)例执解。在此示例中，構(gòu)造函數(shù)傳遞客戶機(jī)標(biāo)識(shí)以另存為實(shí)例數(shù)據(jù)纲酗。 *? ? ? 在回調(diào)中衰腌，將它用來(lái)標(biāo)識(shí)已經(jīng)啟動(dòng)了該回調(diào)的哪個(gè)實(shí)例。 *? ?必須在回調(diào)類中實(shí)現(xiàn)三個(gè)方法： * *http://topmanopensource.iteye.com/blog/1700424 * @author longgangbai */public class MQTTPushCallbackimplements MqttCallback {private static final StringTAG ="PushCallback";? ? private ContextWrappercontext;? ? public JooanMQTTPushCallback(ContextWrapper context) {this.context = context;? ? }/**? ? * 接收到消息的回調(diào)的方法：接收已經(jīng)預(yù)訂的發(fā)布? ? */? ? @Override? ? public void messageArrived(String topic, MqttMessage message)throws Exception {//subscribe（訂閱主題）后得到的消息會(huì)執(zhí)行到這里面//? ? ? ? final NotificationManager notificationManager = (NotificationManager)//? ? ? ? ? ? ? ? context.getSystemService(Context.NOTIFICATION_SERVICE);////? ? ? ? final Notification notification = new Notification(R.drawable.snow,//? ? ? ? ? ? ? ? "Black Ice Warning!", System.currentTimeMillis());////? ? ? ? // Hide the notification after its selected//? ? ? ? notification.flags |= Notification.FLAG_AUTO_CANCEL;////? ? ? ? final Intent intent = new Intent(context, BlackIceActivity.class);//? ? ? ? final PendingIntent activity = PendingIntent.getActivity(context, 0, intent, 0);//? ? ? ? notification.setLatestEventInfo(context, "Black Ice Warning", "Outdoor temperature is " +//? ? ? ? ? ? ? ? new String(message.getPayload()) + "°", activity);//? ? ? ? notification.number += 1;//? ? ? ? notificationManager.notify(0, notification);? ? ? ? Log.w(TAG, "messageArrived: " +"topicName:" + topic +",messageArrived,訂發(fā)送消息失敗: " +new String(message.getPayload()));? ? }/**? ? *? 接收到已經(jīng)發(fā)布的 QoS 1 或 QoS 2 消息的傳遞令牌時(shí)調(diào)用觅赊。? ? * @param token 由 MqttClient.connect 激活此回調(diào)右蕊。? ? */? ? @Override? ? public void deliveryComplete(IMqttDeliveryToken token) {//We do not need this because we do not publish//publish（發(fā)送消息）后會(huì)執(zhí)行到這里? ? ? ? try {? ? ? ? ? ? Log.i("PushCallback", "deliveryComplete: " + token.getMessage().toString());? ? ? ? }catch (MqttException e) {? ? ? ? ? ? e.printStackTrace();? ? ? ? }try {? ? ? ? ? ? Log.i(TAG,"Delivery token \"" + token.hashCode());? ? ? ? ? ? System.out.println("訂閱主題: " + token.getMessageId());? ? ? ? ? ? System.out.println("消息數(shù)據(jù): " + token.getTopics().toString());? ? ? ? ? ? System.out.println("消息級(jí)別(0,1,2): " + token.getGrantedQos().toString());? ? ? ? ? ? System.out.println("是否是實(shí)時(shí)發(fā)送的消息(false=實(shí)時(shí)，true=服務(wù)器上保留的最后消息): " + token.isComplete());? ? ? ? }catch (Exception e) {? ? ? ? ? ? e.printStackTrace();? ? ? ? }? ? }/**? ? * 當(dāng)客戶機(jī)和broker意外斷開(kāi)時(shí)觸發(fā)? ? * 可以再此處理重新訂閱? ? */? ? @Override? ? public void connectionLost(Throwable cause) {//We should reconnect here? //連接丟失后吮螺，一般在這里面進(jìn)行重連? ? ? ? Log.w(TAG, "connectionLost: " + cause.getMessage());? ? ? ? cause.printStackTrace();? ? }}
activity調(diào)用發(fā)送消息（主題和消息主題與服務(wù)器定義一致）：
public String BROKER_URL = "tcp://192.168.1.151:1883";//broker host服務(wù)器地址
private MqttClient mqttClient;
private String clientId = "qaiot/user/f5c71e03-c324-4b32";
private String TOPIC = "qaiot/server/user/1.0/cn";
@Override
protected void onCreate(Bundle savedInstanceState) {//啟動(dòng)服務(wù)
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_jooan_login);
startService(new Intent(this,MQTTService.class));
}
//發(fā)送消息代碼：
new Thread(new Runnable() {
@Override
public void run() {
LogUtil.i("開(kāi)始登錄");
try {
//創(chuàng)建MqttClient對(duì)象
mqttClient = new MqttClient(JooanApplication.get().BROKER_URL, clientId, new MemoryPersistence());
//MqttClient綁定
mqttClient.setCallback(new JooanMQTTPushCallback(JooanLoginActivity.this));
//? ? ? ? ? ? ? ? ? ? ---------------------------------------------------------------
//? ? ? ? ? ? ? ? ? ? MqttConnectOptions connOpts = new MqttConnectOptions();
//? ? ? ? ? ? ? ? ? ? connOpts.setCleanSession(true);
//? ? ? ? ? ? ? ? ? ? System.out.println("Connecting to broker: "+BROKER_URL);
//? ? ? ? ? ? ? ? ? ? mqttClient.connect(connOpts);
//? ? ? ? ? ? ? ? ? ? System.out.println("Connected");
//? ? ? ? ? ? ? ? ? ? System.out.println("Publishing message: "+"Message from MqttPublishSample");
//? ? ? ? ? ? ? ? ? ? ---------------------------------------------------------------
//MqttClient綁定
mqttClient.connect();
//Subscribe to all subtopics of homeautomation
//? ? ? ? ? ? ? ? ? ? mqttClient.subscribe(TOPIC);
//創(chuàng)建MQTT相關(guān)的主題
MqttTopic temperatureTopic = mqttClient.getTopic(TOPIC);
Gson gson = new Gson();
String toJson = gson.toJson(getData());
Log.e("MQTTService", "gson對(duì)象:" + toJson);
//創(chuàng)建MQTT的消息體
MqttMessage message = new MqttMessage(toJson.getBytes());
//設(shè)置消息傳輸?shù)念愋?消息級(jí)別(0,1,2)
message.setQos(1);
//設(shè)置是否在服務(wù)器中保存消息體
message.setRetained(false);
//設(shè)置消息的內(nèi)容
//? ? ? ? ? ? ? ? ? ? message.setPayload(WSMQTTServerCommon.publication.getBytes());
//發(fā)送消息并獲取回執(zhí)
MqttDeliveryToken token = temperatureTopic.publish(message);//發(fā)送消息
//? ? ? ? ? ? ? ? ? ? token.waitForCompletion();設(shè)置超時(shí)
System.out.println("Publishing \"" + message.toString()
+ "\" on topic \"" + temperatureTopic.getName() + "\" with QoS = "
+ message.getQos());
System.out.println("For client instance \"" + mqttClient.getClientId()
+ "\" on address " + mqttClient.getServerURI() + "\"");
System.out.println("With delivery token \"" + token.hashCode()
+ " delivered: " + token.isComplete());
//關(guān)閉連接
//? ? ? ? ? ? ? ? ? ? if (mqttClient.isConnected())
//? ? ? ? ? ? ? ? ? ? ? ? mqttClient.disconnect(Integer.parseInt(System.getProperty("timeout", "10000")));
//? ? ? ? ? ? ? ? ? ? Log.i(TAG, "發(fā)送消息的超時(shí)時(shí)間: "+Integer.parseInt(System.getProperty("timeout", "10000")));
} catch (MqttException e) {
Toast.makeText(getApplicationContext(), "Something went wrong!" + e.getMessage(), Toast.LENGTH_LONG).show();
e.printStackTrace();
LogUtil.i(e.getMessage());
}
}
}).start();
@Override
protected void onDestroy() {
super.onDestroy();
recycler_view.clearOnScrollListeners();
JooanMQTTService.stopMQTTService();//停止服務(wù)
Intent intent = new Intent(this, MQTTService.class);
stopService(intent);
}
