基于MQTT的消息推送

WindySky 2018-03-20

展开全文

这段时间学习了推送技术，对xmpp和mqtt 协议做了下比较。

xmpp基于xml信息传递，所以传输信息量比较大，在保持长链接情况下功耗会比较大。

可能还是比较适合用来做聊天之类的通讯应用，而对于智能和物联低功耗设备的推送来说，感觉比较笨重。

而mqtt协议就是针对网络带宽低，高延时，通信不稳定的环境设计的，特别适合物联设备。低通讯量连接保持，简约轻便。

提供了发布/订阅模式，只要订阅了，即使发布时客户端离线，等再次上线时还能收到消息。
提供了发布反馈，客户端收到消息反馈等机制。
提供了发布质量，比如至多一次，至少一次，只有一次。可以根据不同业务要求进行选择。
提供了心跳机制，可自行设置心跳
提供了鉴权机制

可见，提供的功能已经很完整了。

mqtt 详细协议可见：

http://docs./mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html

中文版的：

http://wenku.baidu.com/link?url=eyG24CVjmmA_J5KbR_Y8xzDOkq5YsrChshcHxcZQzav2OcVE9KNxukXr1uix1ACPTw36QIKwHRTmLqjM1Bm7qqPwVQYG9BzIv_BdJvPoU8_

现在有很多基于mqtt的开源实现，包含客户端和服务器端(或者说中间件)。

我做了一个demo，客户选用paho 提供client-mqtt 和 android server库，中间件是apollo。

apollo 下载地址:

http://activemq./apollo/download.html

下载完解压按照里面的readme 建立和运行自己的broker,依赖java运行环境。

client-mqtt 和 android server库地址：

http:///artifact/org.eclipse.paho

https://www./java/org.eclipse.paho:org.eclipse.paho.client.mqttv3/1.1.0

可以直接把jar文件导入工程，但是我选择的是把源码放入，因为这样可以根据自己的需要对源码做修改。

在android studio上：

1.把jar放入很简单：放到model libs目录下，右键 As libraries就可以了。

2. 源码放入，把sources.jar 分别解压，放到对应java目录下。

3. client.mqtt3 里包含了java的.properties 文件，它们是针对多语言的，这点和android的设计不一样，所以android studio编译时是不能直接把它们打包到class文件里面的。需要单独把它们打包成jar放到lib目录下，这样在运行过程就不会报找不到它们了。

打包很简单，新建org\eclipse\paho\client\mqttv3\internal\nls 目录，把它们都放到这个目录下。

然后在org 目录外执行 jar -cvf properties.jar ./* ，之后把properties.jar 放入libs下。

4. 在androidManifest 里面添加

<service android:name="org.eclipse.paho.android.service.MqttService">
</service>

5. 在自己的activity或者service里面通过调用paho.android.service MqttAndroidClient 类实现发布和订阅。

[java] view plain copy

print?

package com.tww.test;
import android.content.ComponentName;
import android.content.Context;
import android.content.ServiceConnection;
import android.os.Bundle;
import android.os.IBinder;
import android.os.PowerManager;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.android.service.MqttTraceHandler;
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.MqttCallback;
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.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
import java.io.IOException;
import java.io.InputStream;
import javax.net.ssl.SSLSocketFactory;
public class MainActivity extends AppCompatActivity implements View.OnClickListener,ServiceConnection, MqttCallback, IMqttActionListener,MqttTraceHandler {
private static final String TAG = "MainActivity";
private Button mButton;
private MqttAndroidClient mMqttAndroidClient;
private String host = "tcp://192.168.43.224:61613";
private String userName = "admin";
private String passWord = "password";
private IMqttToken mConnectToken;
private PowerManager.WakeLock mWakeLock;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
mButton = (Button) findViewById(R.id.button);
mButton.setOnClickListener(this);
mMqttAndroidClient = new MqttAndroidClient(this,host,"123456789",new MqttDefaultFilePersistence(getCacheDir().getAbsolutePath()));
mMqttAndroidClient.setCallback(this);
mMqttAndroidClient.setTraceEnabled(true);
mMqttAndroidClient.setTraceCallback( this);
Log.d(TAG,"onCreate");
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(true);
options.setUserName(userName);
options.setPassword(passWord.toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(2*60);
options.setMqttVersion(3);
try {
InputStream caInput = getResources().getAssets().open("keystore.bks");
if(caInput!=null){
Log.d(TAG,"do setSocketFactory");
SSLSocketFactory sslSocketFactory = mMqttAndroidClient.getSSLSocketFactory(caInput,"password");
options.setSocketFactory(sslSocketFactory);
}
} catch (IOException e) {
e.printStackTrace();
Log.e(TAG,"do connect IOException:"+e);
}catch (MqttSecurityException e) {
e.printStackTrace();
Log.e(TAG,"do connect MqttSecurityException:"+e);
}
try {
Log.d(TAG,"do connect");
mConnectToken = mMqttAndroidClient.connect(options,this);
} catch (MqttException e) {
Log.e(TAG,"connect MqttException:"+e);
e.printStackTrace();
}
PowerManager powerManager = (PowerManager)getSystemService(Context.POWER_SERVICE);
mWakeLock = powerManager.newWakeLock(PowerManager.FULL_WAKE_LOCK, TAG);
mWakeLock.setReferenceCounted(false);
mWakeLock.acquire();
}
@Override
protected void onDestroy(){
super.onDestroy();
if(mMqttAndroidClient != null){
try {
mMqttAndroidClient.disconnect();
} catch (MqttException e) {
e.printStackTrace();
}
mMqttAndroidClient.close();
}
mWakeLock.release();
}
@Override
public void onClick(View v) {
if(v.getId()==R.id.button){
}
}
@Override
public void onServiceConnected(ComponentName name, IBinder service) {
Log.d(TAG,"onServiceConnected");
}
@Override
public void onServiceDisconnected(ComponentName name) {
Log.d(TAG,"onServiceDisconnected");
}
@Override
public void connectionLost(Throwable throwable) {
Log.d(TAG,"connectionLost");
}
@Override
public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
Log.d(TAG,"messageArrived:"+mqttMessage.toString());
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
Log.d(TAG,"deliveryComplete");
}
@Override
public void onSuccess(IMqttToken iMqttToken) {
if(mConnectToken.equals(iMqttToken)){
try {
Log.d(TAG,"connect success, do subscribe");
mMqttAndroidClient.subscribe("test",1);
} catch (MqttException e) {
e.printStackTrace();
}
}
}
@Override
public void onFailure(IMqttToken iMqttToken, Throwable throwable) {
if(mConnectToken.equals(iMqttToken)){
Log.d(TAG,"onFailure success:"+throwable.toString());
}
}
@Override
public void traceDebug(String s, String s1) {
Log.d(s,s1);
}
@Override
public void traceError(String s, String s1) {
Log.e(s,s1);
}
@Override
public void traceException(String s, String s1, Exception e) {
Log.e(s,s1,e);
}
}