Android(安卓)端实现mqtt消息接收

1为了搞这个MQTT百度收搜了很多资料都是用阿里云的MQTT
虽然好用但是不符合我项目的需求，阿里云的mqtt达到一定的数量就要另行收费。所以公司决定自己搭建服务器实现节省成本

第一步首先后台先搭建一个服务器

主要讲下Android如何使用MQTT通讯。用到的软件或者框架有：

EMQ：https://www.emqx.io/cn/
org.eclipse.paho的MQTT通讯框架：https://github.com/eclipse/paho.mqtt.android

1.安装所需要的依赖包
$ sudo yum install -y yum-utils device-mapper-persistent-data lvm2

2.使用以下命令设置稳定存储库，以 CentOS7 为例
$ sudo yum-config-manager --add-repo https://repos.emqx.io/emqx-ce/redhat/centos/7/emqx-ce.repo

3.安装最新版本的 EMQ X
$ sudo yum install emqx

4.安装特定版本的 EMQ X

$ yum list emqx --showduplicates | sort -r
emqx.x86_64 3.1.0-1.el7 emqx-stable
emqx.x86_64 3.0.1-1.el7 emqx-stable
emqx.x86_64 3.0.0-1.el7 emqx-stable

5我这里使用的是这个版本
$ sudo yum install emqx-3.1.0

6.启动 EMQ X

$ emqx start
emqx 3.1.0 is started successfully!

$ emqx_ctl status
Node ‘emqx@127.0.0.1’ is started
emqx v3.1.0 is running

systemctl 启动
$ sudo systemctl start emqx

service 启动
$ sudo service emqx start

EMQ管理后台

二、Android使用MQTT

1 APP build.gradle下
defaultConfig {
multiDexEnabled true
}

2 app build.gradle下
dependencies {

compile 'com.aliyun.alink.linksdk:iot-linkkit:1.6.6'compile 'com.aliyun.alink.linksdk:breeze:1.2.3'compile 'com.aliyun.alink.linksdk:breeze-biz:1.1.0'compile 'com.aliyun.alink.linksdk:breeze-ota:1.2.1'

}

3 在工程目录下bulid.gradle
allprojects {
repositories {
jcenter()
maven { url ‘https://maven.google.com’ }
google()
maven {
url “http://maven.aliyun.com/nexus/content/repositories/releases/”
}
maven {
url “http://maven.aliyun.com/nexus/content/repositories/snapshots”
}
}
}

4 连接mqtt并订阅

String broker = “tcp://www.xxxo.com:1883”;// 你配置的服务器地址
String clientId = “t3”;//MQTT的客户端ID
//Use the memory persistence
MemoryPersistence persistence = new MemoryPersistence();
try {
// broker为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
// MQTT的连接设置
MqttConnectOptions connOpts = new MqttConnectOptions();
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
connOpts.setCleanSession(true);
// 设置连接的用户名
connOpts.setUserName(“test”);
String password = “11”;
connOpts.setKeepAliveInterval(60);//心跳时间
// 设置超时时间单位为秒
connOpts.setConnectionTimeout(10);
// 设置连接的密码
connOpts.setPassword(password.toCharArray());
// System.out.println(“Connecting to broker:” + broker);
Log.d(TAG, "onCreate:= " + “连接地址” + broker);
sampleClient.connect(connOpts);

        //  System.out.println("Connected");        Log.d(TAG, "onCreate: " + "开始连接" + connOpts);        String topic = "/opendoor";//订阅的主题        //  System.out.println("Subscribe to topic:" + topic);        //   Log.d(TAG, "onCreate:C " +topic);        //订阅消息        sampleClient.subscribe(topic);        // 设置回调        sampleClient.setCallback(new MqttCallback() {            public void messageArrived(String topic, MqttMessage message) throws Exception {                String theMsg = MessageFormat.format("{0} is arrived for topic {1}.", new String(message.getPayload()), topic);                //   System.out.println(theMsg);                Log.d(TAG, "onCreate:接收主题" + topic);                Log.d(TAG, "onCreate=接收消息" + new String(message.getPayload()));            }            public void deliveryComplete(IMqttDeliveryToken token) {            }            public void connectionLost(Throwable throwable) {            }        });        String content = "发送数据  ";        int qos = 1;        // System.out.println("Publishing message:" + content);        //  Log.d(TAG, "onCreate:E " +content);        //  MqttMessage message = new MqttMessage(content.getBytes());        //  message.setQos(qos);        // message.setPayload(content.getBytes());        // sampleClient.publish(topic, message);        // System.out.println("Message published");        // Log.d(TAG, "onCreate: " +content);    } catch (MqttException me) {        // System.out.println("reason" + me.getReasonCode());        Log.d(TAG, "onCreate:=me.getReasonCode() " + me.getReasonCode());        // System.out.println("msg" + me.getMessage());        Log.d(TAG, "onCreate:= me.getMessage()" + me.getMessage());        // System.out.println("loc" + me.getLocalizedMessage());        Log.d(TAG, "onCreate:H = me.getLocalizedMessage()" + me.getLocalizedMessage());        //  System.out.println("cause" + me.getCause());        Log.d(TAG, "onCreate:= me.getCause() " + me.getCause());        //  System.out.println("excep" + me);        Log.d(TAG, "onCreate:=me" + me);        //  me.printStackTrace();    }

第5步：如果需要发布的话

String broker = "tcp://xxx.com:1883";// 1883    String clientId = "t3";    //Use the memory persistence    MemoryPersistence persistence = new MemoryPersistence();    try {        // broker为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存        MqttClient sampleClient = new MqttClient(broker, clientId, persistence);        // MQTT的连接设置        MqttConnectOptions connOpts = new MqttConnectOptions();        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接        connOpts.setCleanSession(true);        // 设置连接的用户名        connOpts.setUserName("test");        String password = "11";        connOpts.setKeepAliveInterval(60);//心跳时间        // 设置超时时间 单位为秒        connOpts.setConnectionTimeout(10);        // 设置连接的密码        connOpts.setPassword(password.toCharArray());        //   System.out.println("Connecting to broker:" + broker);        Log.d(TAG, "onCreate:= " + "连接地址" + broker);        sampleClient.connect(connOpts);        //  System.out.println("Connected");        Log.d(TAG, "onCreate: " + "开始连接" + connOpts);        String topic = "/opendoor";//        //  System.out.println("Subscribe to topic:" + topic);        String content = "发送数据 是否能OA判断上个了个网格为奇偶器ngo琥珀黄SUFPDFOAJ  ";        int qos = 1;        // System.out.println("Publishing message:" + content);        //  Log.d(TAG, "onCreate:E " +content);        MqttMessage message = new MqttMessage(content.getBytes());        //  message.setQos(qos);        message.setPayload(content.getBytes());        sampleClient.publish(topic, message);        // System.out.println("Message published");        Log.d(TAG, "onCreate: " + content);    } catch (MqttSecurityException e) {        e.printStackTrace();    } catch (MqttException e) {        e.printStackTrace();    }

更多相关文章

随机推荐