mqtt android封装,Android之MQTT封装使用
一、MQTT協(xié)議介紹
客戶機較小并且 MQTT 協(xié)議高效地使用網(wǎng)絡(luò)帶寬，在這個意義上，其為輕量級。MQTT 協(xié)議支持可靠的傳送和即發(fā)即棄的傳輸。 在此協(xié)議中，消息傳送與應(yīng)用程序脫離。 脫離應(yīng)用程序的程度取決于寫入 MQTT 客戶機和 MQTT 服務(wù)器的方式。脫離式傳送能夠?qū)?yīng)用程序從任何服務(wù)器連接和等待消息中解脫出來。 交互模式與電子郵件相似，但在應(yīng)用程序編程方面進行了優(yōu)化。
二、MQTT優(yōu)點
1.一對多消息分發(fā)外，發(fā)布/預(yù)訂也脫離了應(yīng)用程序。對于具有多個客戶機的應(yīng)用程序來說，這些功能非常有用。
2.它與消息內(nèi)容沒有任何關(guān)系。
3.它通過 TCP/IP 運行，TCP/IP 可以提供基本網(wǎng)絡(luò)連接。
4.它針對消息傳送提供三種服務(wù)質(zhì)量：
“至多一次”
消息根據(jù)底層因特網(wǎng)協(xié)議網(wǎng)絡(luò)盡最大努力進行傳遞。 可能會丟失消息。
例如，將此服務(wù)質(zhì)量與通信環(huán)境傳感器數(shù)據(jù)一起使用。 對于是否丟失個別讀取或是否稍后立即發(fā)布新的讀取并不重要。
“至少一次”
保證消息抵達，但可能會出現(xiàn)重復(fù)。
“剛好一次”
確保只收到一次消息。
例如，將此服務(wù)質(zhì)量與記帳系統(tǒng)一起使用。
重復(fù)或丟失消息可能會導(dǎo)致不便或收取錯誤費用。
三、相關(guān)資料
MQTT官網(wǎng)：http://mqtt.org/
MQTT介紹：http://www.ibm.com
MQTT Android github：https://github.com/eclipse/paho.mqtt.android
MQTT API：http://www.eclipse.org/paho/files/javadoc/index.html
MQTT Android API： http://www.eclipse.org/paho/files/android-javadoc/index.html
四、?MQTT Android客戶端具體實現(xiàn)
基本概念：
topic：中文意思是“話題”。在MQTT中訂閱了(subscribe)同一話題(topic)的客戶端會同時收到消息推送。直接實現(xiàn)了“群聊”功能。
clientId：客戶身份唯一標識。
qos：服務(wù)質(zhì)量。
retained：要保留最后的斷開連接信息。
MqttAndroidClient#subscribe()：訂閱某個話題。
MqttAndroidClient#publish()： 向某個話題發(fā)送消息，之后服務(wù)器會推送給所有訂閱了此話題的客戶。
userName：連接到MQTT服務(wù)器的用戶名。
passWord ：連接到MQTT服務(wù)器的密碼。
添加依賴
repositories {
maven {
url "https://repo.eclipse.org/content/repositories/paho-releases/"
}
}
dependencies {
compile 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'
compile 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.0'
}
添加限權(quán)
注冊
Service
創(chuàng)建service
public class MQTTService extends Service {
public static final String TAG = MQTTService.class.getSimpleName();
private static MqttAndroidClient client;
private MqttConnectOptions conOpt;
private String host = "tcp://192.168.229.80:1883";
private String userName = "admin";
private String passWord = "password";
private static String myTopic = "gmt/report/" + UserConfig.get().getAccount();//要訂閱的主題
private static String sendTopic = "gmt/control/" + UserConfig.get().getmGateway();//要發(fā)布控制主題
private String clientId = "152xxxx2901";//客戶端標識
private IGetMessageCallBack mGetMessageCallBack;
static IGetMessageCallBack mcallBack;
@Override
public void onCreate() {
super.onCreate();
Log.e(getClass().getName(), "onCreate");
init();
}
public static void publish(String msg) {
String topic = sendTopic;
Integer qos = 0;
Boolean retained = false;
try {
if (client != null) {
client.publish(topic, msg.getBytes(), qos.intValue(), retained.booleanValue());
}
} catch (MqttException e) {
e.printStackTrace();
}
}
public static void publish(String msg, IGetMessageCallBack callBack) {
mcallBack = callBack;
String topic = sendTopic;
Integer qos = 0;
Boolean retained = false;
if (client != null) {
try {
client.publish(topic, msg.getBytes(), qos.intValue(), retained.booleanValue());
} catch (MqttException e) {
e.printStackTrace();
}
}
}
private void init() {
// 服務(wù)器地址(協(xié)議+地址+端口號)
String uri = host;
client = new MqttAndroidClient(this, uri, clientId);
// 設(shè)置MQTT監(jiān)聽并且接受消息
client.setCallback(mqttCallback);
conOpt = new MqttConnectOptions();
// 清除緩存
conOpt.setCleanSession(true);
// 設(shè)置超時時間，單位：秒
conOpt.setConnectionTimeout(10);
// 心跳包發(fā)送間隔，單位：秒
conOpt.setKeepAliveInterval(20);
// 用戶名
conOpt.setUserName(userName);
// 密碼
conOpt.setPassword(passWord.toCharArray()); //將字符串轉(zhuǎn)換為字符串數(shù)組
// last will message
boolean doConnect = true;
String message = "{\"terminal_uid\":\"" + clientId + "\"}";
Log.e(getClass().getName(), "message是:" + message);
String topic = myTopic;
Integer qos = 0;
Boolean retained = false;
if ((!message.equals("")) || (!topic.equals(""))) {
// 最后的遺囑
// MQTT本身就是為信號不穩(wěn)定的網(wǎng)絡(luò)設(shè)計的，所以難免一些客戶端會無故的和Broker斷開連接。
//當客戶端連接到Broker時，可以指定LWT，Broker會定期檢測客戶端是否有異常。
//當客戶端異常掉線時，Broker就往連接時指定的topic里推送當時指定的LWT消息。
try {
conOpt.setWill(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
} catch (Exception e) {
Log.i(TAG, "Exception Occured", e);
doConnect = false;
iMqttActionListener.onFailure(null, e);
}
}
if (doConnect) {
doClientConnection();
}
}
@Override
public void onDestroy() {
// 此處解決MQTT退出異常問題
if (client != null) {
client.unregisterResources();
client.close();
}
super.onDestroy();
}
/**
* 連接MQTT服務(wù)器
*/
private void doClientConnection() {
if (!client.isConnected() && isConnectIsNormal()) {
try {
client.connect(conOpt, null, iMqttActionListener);
} catch (MqttException e) {
e.printStackTrace();
}
}
}
// MQTT是否連接成功
private IMqttActionListener iMqttActionListener = new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken arg0) {
Log.i(TAG, "連接成功 ");
try {
// 訂閱myTopic話題
Log.i(TAG, "SUB:" + myTopic);
client.subscribe(myTopic, 0);
} catch (MqttException e) {
e.printStackTrace();
Log.e(TAG, "sube：" + e.getMessage());
}
}
@Override
public void onFailure(IMqttToken arg0, Throwable arg1) {
arg1.printStackTrace();
// 連接失敗，重連
}
};
// MQTT監(jiān)聽并且接受消息
private MqttCallback mqttCallback = new MqttCallback() {
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
String str1 = new String(message.getPayload());
JSONObject jsonObject = new JSONObject(str1);
JSONObject msg = jsonObject.optJSONObject("msg");
if (mGetMessageCallBack != null && msg != null) {
mGetMessageCallBack.setMessage(msg);
mcallBack.setMessage(msg);
}
String str2 = topic + ";qos:" + message.getQos() + ";retained:" + message.isRetained();
Log.i(TAG, "messageArrived:" + str1);
Log.i(TAG, str2);
}
@Override
public void deliveryComplete(IMqttDeliveryToken arg0) {
}
@Override
public void connectionLost(Throwable arg0) {
// 失去連接，重連
}
};
/**
* 判斷網(wǎng)絡(luò)是否連接
*/
private boolean isConnectIsNormal() {
ConnectivityManager connectivityManager = (ConnectivityManager) this.getApplicationContext()
.getSystemService(Context.CONNECTIVITY_SERVICE);
NetworkInfo info = connectivityManager.getActiveNetworkInfo();
if (info != null && info.isAvailable()) {
String name = info.getTypeName();
Log.i(TAG, "MQTT當前網(wǎng)絡(luò)名稱：" + name);
return true;
} else {
Log.i(TAG, "MQTT 沒有可用網(wǎng)絡(luò)");
return false;
}
}
@Override
public IBinder onBind(Intent intent) {
Log.e(getClass().getName(), "onBind");
return new CustomBinder();
}
public void setIGetMessageCallBack(IGetMessageCallBack IGetMessageCallBack) {
this.mGetMessageCallBack = IGetMessageCallBack;
}
public class CustomBinder extends Binder {
public MQTTService getService() {
return MQTTService.this;
}
}
@Override
public boolean onUnbind(Intent intent) {
return super.onUnbind(intent);
}
}
接口回調(diào)
public interface IGetMessageCallBack {
public void setMessage(JSONObject message);
}
代碼使用
public class MainActivity extends BaseActivity implements IGetMessageCallBack {
@Override
protected int initLayout() {
return R.layout.activity_main;
}
@Override
protected void initView() {
initBind();
// 此處注冊方便回調(diào)
getServiceConnection().setIGetMessageCallBack(this);
}
@Override
protected void onDestroy() {
super.onDestroy();
}
// 此處接收消息
@Override
public void setMessage(JSONObject message) {
Log.e("bingo", "main:" + message);
}
}
注意要在BaseActivity中封裝如下
public abstract class BaseActivity extends AppCompatActivity {
private Context mContext;
private Handler handler;
private MyServiceConnection serviceConnection;
public MyServiceConnection getServiceConnection() {
return serviceConnection;
}
@Override
public void onCreate(@Nullable Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(initLayout());
mContext = this;
initConnectTion();
}
// 此處申請綁定
private void initConnectTion() {
serviceConnection = new MyServiceConnection();
Intent intent = new Intent(this, MQTTService.class);
bindService(intent, serviceConnection, Context.BIND_AUTO_CREATE);
}
@Override
protected void onResume() {
super.onResume();
}
public Context getContext() {
return mContext;
}
@Override
protected void onDestroy() {
// 要在此處進行解綁
if (serviceConnection != null) {
unbindService(serviceConnection);
}
super.onDestroy();
}
}
異常
總結(jié)
以上是生活随笔為你收集整理的mqtt android封装,Android之MQTT封装使用的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 初识卷积神经网络
- 下一篇： android sina oauth2.