Android mqtt入門 Android studio

2018年04月09日 14:02:30?hbw020?閱讀數：1564

分享 mqtt簡單使用介紹：

1、as創建工程

2、官網下載mqtt支持包放入lib文件，點擊打開鏈接，https://repo.eclipse.org/content/repositories/paho-releases/org/eclipse/paho/org.eclipse.paho.client.mqttv3/1.2.0/。當前使用的是：

org.eclipse.paho.client.mqttv3-1.2.0.jar

3、然后開始工作：

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;/***全局主服務分發和接收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即時推送", 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收到推送結果: " + (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(), "連接失敗，系統正在重連", Toast.LENGTH_SHORT).show();? ? ? ? ? ? ? ? ? ? Log.i(TAG, "連接失敗，系統正在重連");? ? ? ? ? ? ? ? }? ? ? ? ? ? }? ? ? ? };? ? ? ? startReconnect();? ? }private void initMqttClient() {try {//host為主機名，test為clientid即連接MQTT的客戶端ID，一般以客戶端唯一標識符表示，MemoryPersistence設置clientid的保存形式，默認為以內存保存? ? ? ? ? ? client=new MqttClient(JooanApplication.get().BROKER_URL, myTopic, new MemoryPersistence());? ? ? ? ? ? //MQTT的連接設置? ? ? ? ? ? options =new MqttConnectOptions();? ? ? ? ? ? //設置是否清空session,這里如果設置為false表示服務器會保留客戶端的連接記錄，這里設置為true表示每次連接到服務器都以新的身份連接? ? ? ? ? ? options.setCleanSession(true);? ? ? ? ? ? //設置連接的用戶名? ? ? ? ? ? options.setUserName(JooanApplication.get().userName);? ? ? ? ? ? //設置連接的密碼? ? ? ? ? ? options.setPassword(JooanApplication.get().passWord.toCharArray());? ? ? ? ? ? // 設置超時時間 單位為秒? ? ? ? ? ? options.setConnectionTimeout(10);? ? ? ? ? ? // 設置會話心跳時間 單位為秒 服務器會每隔1.5*20秒的時間向客戶端發送個消息判斷客戶端是否在線，但這個方法并沒有重連的機制? ? ? ? ? ? options.setKeepAliveInterval(20);//? ? ? ? ? ? options.setWill();//如果項目中需要知道客戶端是否掉線可以調用該方法? ? ? ? ? ? //設置回調? ? ? ? ? ? client.setCallback(new MqttCallback() {@Override? ? ? ? ? ? ? ? public void connectionLost(Throwable cause) {//連接丟失后，一般在這里面進行重連? ? ? ? ? ? ? ? ? ? Log.w(TAG, "connectionLost----------: " + cause.getMessage());? ? ? ? ? ? ? ? }@Override? ? ? ? ? ? ? ? public void messageArrived(String topic, MqttMessage message)throws Exception {//subscribe后得到的消息會執行到這里面? ? ? ? ? ? ? ? ? ? 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后會執行到這里? ? ? ? ? ? ? ? ? ? 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();? ? }//? ? 判斷服務是否運行中? ? 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;/***發布消息的回調類**必須實現MqttCallback的接口并實現對應的相關接口方法*? ? ? ?CallBack類將實現MqttCallBack。每個客戶機標識都需要一個回調實例。在此示例中，構造函數傳遞客戶機標識以另存為實例數據。*在回調中，將它用來標識已經啟動了該回調的哪個實例。*? ?必須在回調類中實現三個方法：**http://topmanopensource.iteye.com/blog/1700424*@authorlonggangbai*/public class MQTTPushCallbackimplements MqttCallback {private static final StringTAG="PushCallback";? ? private ContextWrappercontext;? ? public JooanMQTTPushCallback(ContextWrapper context) {this.context = context;? ? }/***接收到消息的回調的方法：接收已經預訂的發布*/@Override? ? public void messageArrived(String topic, MqttMessage message)throws Exception {//subscribe(訂閱主題)后得到的消息會執行到這里面//? ? ? ? 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,訂發送消息失敗: " +new String(message.getPayload()));? ? }/***接收到已經發布的QoS 1或QoS 2消息的傳遞令牌時調用。*@paramtoken由MqttClient.connect激活此回調。*/@Override? ? public void deliveryComplete(IMqttDeliveryToken token) {//We do not need this because we do not publish//publish(發送消息)后會執行到這里? ? ? ? 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("消息數據: " + token.getTopics().toString());? ? ? ? ? ? System.out.println("消息級別(0,1,2): " + token.getGrantedQos().toString());? ? ? ? ? ? System.out.println("是否是實時發送的消息(false=實時，true=服務器上保留的最后消息): " + token.isComplete());? ? ? ? }catch (Exception e) {? ? ? ? ? ? e.printStackTrace();? ? ? ? }? ? }/***當客戶機和broker意外斷開時觸發*可以再此處理重新訂閱*/@Override? ? public void connectionLost(Throwable cause) {//We should reconnect here? //連接丟失后，一般在這里面進行重連? ? ? ? Log.w(TAG, "connectionLost: " + cause.getMessage());? ? ? ? cause.printStackTrace();? ? }}

activity調用發送消息(主題和消息主題與服務器定義一致)：

public String BROKER_URL = "tcp://192.168.1.151:1883";//broker host服務器地址

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) {//啟動服務

super.onCreate(savedInstanceState);

setContentView(R.layout.activity_jooan_login);

startService(new Intent(this,MQTTService.class));

}

//發送消息代碼：

new Thread(new Runnable() {

@Override

public void run() {

LogUtil.i("開始登錄");

try {

//創建MqttClient對象

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);

//創建MQTT相關的主題

MqttTopic temperatureTopic = mqttClient.getTopic(TOPIC);

Gson gson = new Gson();

String toJson = gson.toJson(getData());

Log.e("MQTTService", "gson對象:" + toJson);

//創建MQTT的消息體

MqttMessage message = new MqttMessage(toJson.getBytes());

//設置消息傳輸的類型:消息級別(0,1,2)

message.setQos(1);

//設置是否在服務器中保存消息體

message.setRetained(false);

//設置消息的內容

//? ? ? ? ? ? ? ? ? ? message.setPayload(WSMQTTServerCommon.publication.getBytes());

//發送消息并獲取回執

MqttDeliveryToken token = temperatureTopic.publish(message);//發送消息

//? ? ? ? ? ? ? ? ? ? token.waitForCompletion();設置超時

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());

//關閉連接

//? ? ? ? ? ? ? ? ? ? if (mqttClient.isConnected())

//? ? ? ? ? ? ? ? ? ? ? ? mqttClient.disconnect(Integer.parseInt(System.getProperty("timeout", "10000")));

//? ? ? ? ? ? ? ? ? ? Log.i(TAG, "發送消息的超時時間: "+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();//停止服務

Intent intent = new Intent(this, MQTTService.class);

stopService(intent);

}