Android 对接OneNet Mqtt服务Demo_android 与onenet通信

一、开通OneNet Mqtt

官网：https://open.iot.10086.cn/
在控制台首页选择多协议接入，然后添加产品，设备接入协议选择MQTT，再然后创建设备，查看设备详情。

产品ID、设备ID、鉴权信息等下会用到

二、开始编程

1、添加依赖、声明权限

创建一个新android项目，添加以下依赖，声明权限

    implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'
    implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.1'
1
2

    <uses-permission android:name="android.permission.WAKE_LOCK" />
    <uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE" />
    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
    <uses-permission android:name="android.permission.READ_PHONE_STATE" />
    <uses-permission android:name="android.permission.READ_EXTERNAL_STORAGE" />
    <uses-permission android:name="android.permission.INTERNET" />`
1
2
3
4
5
6

同时注册Mqtt服务service：

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

2、MainActivity.java全部代码

看代码和注释吧，代码不多，不难理解。
当向某个topic发布消息时，所有订阅了该topic的设备都将收到消息。
在这个Demo中，点击订阅然后再发布消息，也会收到自己发布的消息，因为这个当前设备也订阅了topic。

package com.example.mqttdemo;

import androidx.appcompat.app.AppCompatActivity;

import android.annotation.SuppressLint;
import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.Button;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
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;

public class MainActivity extends AppCompatActivity {

    private static final String TAG = "MainActivity";
    private MqttAndroidClient client;
    private MqttCallback mqttCallback;
    private MqttConnectOptions conOpt;

    private static final String SERVER_URL = "tcp://mqtt.heclouds.com:6002";
    private static final String CLIENT_ID = "你的设备ID";
    private static final String USER_NAME = "你的产品ID";
    private static final String PASSWORD = "设备鉴权信息";
    private Button mConnBtn;
    private IMqttActionListener mqttActionListener;
    private Button mSubscribeBtn;
    private Button mPublishBtn;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        initData();
        initView();
        //开始初始化MQTT
        client = new MqttAndroidClient(this, SERVER_URL, CLIENT_ID);
        // 设置MQTT监听并且接受消息
        client.setCallback(mqttCallback);
        conOpt = new MqttConnectOptions();
        // 清除缓存
        conOpt.setCleanSession(true);
        // 设置超时时间，单位：秒
        conOpt.setConnectionTimeout(10);
        // 心跳包发送间隔，单位：秒
        conOpt.setKeepAliveInterval(30);
        // 用户名
        conOpt.setUserName(USER_NAME.trim());
        // 密码
        conOpt.setPassword(PASSWORD.trim().toCharArray());

    }

    /**
     * 创建一些监听器
     */
    private void initData() {
        mqttCallback = new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {

            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                //收到消息，打印消息
                String msg = new String(message.getPayload());
                Log.d(TAG,"收到消息：" + msg);
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {

            }
        };
        //mqtt连接结果监听
        mqttActionListener = new IMqttActionListener() {
            @Override
            public void onSuccess(IMqttToken asyncActionToken) {
                Log.d(TAG,"连接成功");
            }

            @Override
            public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                Log.d(TAG,"连接失败");
            }
        };
    }

    private void initView() {
        mConnBtn = findViewById(R.id.connClientBtn);
        mSubscribeBtn = findViewById(R.id.SubscribeBtn);
        mPublishBtn = findViewById(R.id.publishBtn);
        //发起连接按钮点击
        mConnBtn.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                if (!client.isConnected()) {
                    try {
                        //发起连接
                        client.connect(conOpt, null, mqttActionListener);
                    } catch (MqttException e) {
                        Log.d(TAG, "mqttConnect fail:" + e.getStackTrace());
                    }
                }
            }
        });
        //订阅按钮
        mSubscribeBtn.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                mqttSubscribe("Topic名称");
            }
        });
        //发布消息按钮
        mPublishBtn.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                mqttPublish("Topic名称","数据");
            }
        });
    }

    /**
     * 订阅Topic
     * @param topic
     */
    public void mqttSubscribe(String topic) {
        try {
            // 订阅myTopic话题
            client.subscribe(topic, 1, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    Log.d(TAG,"订阅成功");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    Log.d(TAG,"订阅失败");
                }
            });
        } catch (MqttException e) {
            Log.d(TAG, "MQTT订阅失败:" + e.getStackTrace());
        }
    }

    /**
     * 发布消息
     * @param topic
     * @param msg
     */
    public void mqttPublish(String topic,String msg) {
        Integer qos = 0;
        Boolean retained = false;
        try {
            if (client != null) {
                client.publish(topic, msg.getBytes(), qos.intValue(), retained.booleanValue(), null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        Log.d(TAG,"发布消息成功");
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        Log.d(TAG,"发布消息失败");
                    }
                });
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180