MQTT协议通过Apache|MQTT协议通过Apache Apollo代理服务器，使用eclipse paho发送推送消息到android客户端
前言
MQTT是轻量级基于代理的发布/订阅的消息传输协议，设计思想是开放、简单、轻量、易于实现。这些特点使它适用于受限环境。关于MQTT的详细介绍推荐这篇文章推荐文章
Apache apollo是一个代理服务器，其是在ActiveMQ基础上发展而来的，可以支持STOMP, AMQP, MQTT, Openwire, SSL, and WebSockets 等多种协议。运行apollo需要先配置JAVA_HOME
Apollo下载地址
paho是eclipse提供一个访问MQTT服务器的一种开源客户端库，其中提供7种不同平台的客户端类库，今天咱们以java客户端为例子演示
paho下载地址
【MQTT协议通过Apache|MQTT协议通过Apache Apollo代理服务器，使用eclipse paho发送推送消息到android客户端】apollo使用的时候需要 配置JAVA_HOME配置成功之后，命令行指定到bin目录下，然后输入apollo create mybroker（图1）这时候会在bin目录下生成一个mybroker文件夹（图2）在mybroker中进入bin目录（图3）命令行 cd mybroker/bin(图4)然后输入apollo-broker run(图5)显示成功后打开浏览器输入 http://127.0.0.1:61680 这时候要是显示(图6)界面就证明成功了。用户名和密码可以通过myorker/etc 下的users.properties文件中找到默认用户名和密码是 admin和password.
文章图片
1.png 
文章图片
2.png 
文章图片
3.png 
文章图片
4.png 
文章图片
5.png 
文章图片
6.png paho下载完成后进入文件夹启动paho.exe(图7)点击+号创建服务器连接（图8）。服务器地址在之前创建的mybroker/etc文件夹中打开apollo.xml找到图9中那一行改成自己本机的局域网IP地址。完成后在图8中服务器地址中输入相同的地址和端口，这时候点击连接会提示“连接失败-错误的用户名和密码”。在选项中需要开启登录输入之前apollo配置的用户名和密码（图10）重新点击连接这时候连接状态会显示已连接（图11）
文章图片
7.png 
文章图片
8.png 
文章图片
9.png 
文章图片
10.png 
文章图片
11.png 下面是android客户端配置
- module的build.gradle中找到dependencies 然后添加
compile'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.1'compile'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'后Sync Project with Gradle Files
- AndroidManifest.xml中添加
- 初始化MQTT
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.100.41:61613";
private String userName = "admin";
private String passWord = "password";
private static String myTopic = "test";
private String clientId = "clientId";
@Override
public int onStartCommand(Intent intent, int flags, int startId) {
init();
return super.onStartCommand(intent, flags, startId);
}public static void publish(String msg){
Integer qos = 0;
Boolean retained = false;
try {
client.publish(myTopic, msg.getBytes(), qos.intValue(), retained.booleanValue());
} catch (MqttException e) {
e.printStackTrace();
}
}private void init() {
// 服务器地址（协议+地址+端口号）
client = new MqttAndroidClient(this, host, clientId);
// 设置MQTT监听并且接受消息
client.setCallback(mqttCallback);
conOpt = new MqttConnectOptions();
// 清除缓存
conOpt.setCleanSession(true);
// 设置超时时间，单位：秒
conOpt.setConnectionTimeout(10);
// 心跳包发送间隔，单位：秒
conOpt.setKeepAliveInterval(20);
// 用户名
conOpt.setUserName(userName);
// 密码
conOpt.setPassword(passWord.toCharArray());
boolean doConnect = true;
String message = "{\"terminal_uid\":\"" + clientId + "\"}";
Integer qos = 0;
Boolean retained = false;
client.getClientId();
if ((!message.equals("")) || (!myTopic.equals(""))) {
try {
//setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
conOpt.setWill(myTopic, message.getBytes(), qos.intValue(), retained.booleanValue());
} catch (Exception e) {
doConnect = false;
iMqttActionListener.onFailure(null, e);
}
}if (doConnect) {
doClientConnection();
}}@Override
public void onDestroy() {
try {
client.disconnect();
} catch (MqttException e) {
e.printStackTrace();
}
super.onDestroy();
}/** 连接MQTT服务器 */
private void doClientConnection() {
if (!client.isConnected() && isConnectIsNomarl()) {
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话题
client.subscribe(myTopic,1);
} catch (MqttException e) {
e.printStackTrace();
}
}@Override
public void onFailure(IMqttToken arg0, Throwable arg1) {
arg1.printStackTrace();
// 连接失败，重连
Log.i(TAG, "连接失败 ");
}
};
// MQTT监听并且接受消息
private MqttCallback mqttCallback = new MqttCallback() {@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {String str1 = new String(message.getPayload());
MQTTMessage msg = new MQTTMessage();
msg.setMessage(str1);
EventBus.getDefault().post(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) {
// 失去连接，重连
Toast.makeText(MQTTService.this, " 失去连接，重连", Toast.LENGTH_SHORT).show();
}
};
/** 判断网络是否连接 */
private boolean isConnectIsNomarl() {
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当前网络名称：" + name);
return true;
} else {
Log.i(TAG, "MQTT 没有可用网络");
return false;
}
}@Override
public IBinder onBind(Intent intent) {
return null;
}
}最后通过Paho发布消息 客户端就能收到推送过来的内容了！
推荐阅读
- gitlab|gitlab 通过备份还原 admin/runner 500 Internal Server Error
- whlie循环和for循环的应用
- 如何通过锻炼的方法治疗前列腺肥大
- 通过复盘快速成长（附模板）
- （八）签协议
- MyBatis|MyBatis Generator配置
- 运维|如何限制IP 通过 SSH连接服务器
- 运维|Linux 禁止用户或 IP通过 SSH 登录
- 计算机网络|计算机网络——DHCP协议详解
- 青椒板书--足球