引言
本篇博客主要介绍mqtt和emqx配置连接实现数据收发，我会从基础的本机连接到手机和本机连接再到手机实现mqtt连接云平台，大家可以根据需要自行选择观看（后面两个教程都建立在mqtt和emqx下载完成的基础上，若没有下载完成，请不要跳过第一个教程）。
一、电脑本机实现连接
1.EMQX下载
首先，我们要先下载emqx，Directory listing for EMQX: /v5.3.2/ | EMQ，挑选适合版本即可，我这里使用的是emqx5.3.2版本，如下点击链接，下载Windows版本。
下载完成后解压该文件在D盘，注意在英文路径中。然后win+R输入cmd打开终端，进入刚刚解压的文件内容。
或者直接进入该文件夹bin下，在搜索位置输入cmd
（必须以管理员身份） 执行该指令 .\emqx.cmd install 出现下图状态表示该服务成功。该操作执行一次即可。
然后启动emqx： .\emqx.cmd console 出现 EMQX 5.3.2 is running now！证明启动成功。
大家可能会出现无法连接的问题，如该问题为端口号(4370)被占，可以用以下方法解决。(我这里占用4370的程序代号为17080，大家看自己情况写自己查询出来的程序：代码第一行为查询，第二行为执行终止操作。)
若无法终止，则以管理员身份运行再次执行该操作。
//若显示无法连接端口号被占：（终止占用端口4370的程序17080 ）
netstat -aon | findstr 4370
taskkill /PID 17080 /F浏览器输入 localhost:18083 进入EMQX的控制台界面，初始登录账号密码均为：账号：admin，密码：public 。成功登录后，界面会出现让你修改密码的情况，成功修改后进入以下界面。
2.MQTT下载
下载mqttx客户端，Directory listing for MQTTX: /v1.9.7/ | EMQ (emqx.com)
得到安装包后，双击安装mqttx客户端，位置放在D盘，其余默认即可。打开该软件，我们先找到设置，将语言改为中文。
3.mqtt和emqx通信收发信息
打开mqttx，点击 “ + ” 新建连接，名称随便输入，Client ID默认即可，服务器地址为刚才创建的MQTT本地服务器，因为客户端在同一台电脑上，所以地址为127.0.0.1，端口号默认1883，然后点击右上角连接。
我们可以发现emqx终端显示了我们的连接。
然后我们回到mqttx，我们在刚刚创建连接下面添加订阅，主题自定义即可。
添加主题后我们可以在emqx订阅管理中查看到你的主题。
然后回到mqtt，创建成功后我们可以先给这个主题发送消息，看一下是否能收到，如下，在发送消息行输入主题名：mk123（大家输入自己创建的即可）。
OK，我们打开emqx中webSocket部分，点击连接。
订阅主题，自定义即可。
我们向mqtt订阅的主题发送消息，我这里是mk123，可以看到已发送部分出现发送内容，回到mqtt可以看到发送过来的信息 123。
然后我们在mqtt发送，主题为mk456，如下，然后我们回到emqx就能发现emqx已接收部分出现提示收到了456。
综上所述，我们完成了简单的本机mqtt和emqx数据收发！
二、AndroidStudio连接本机电脑
1. 我们首先下载相关jar包---mqttv3和service（可以从Maven库中下载）。
1.下载jar包---mqttv3和service
Central Repository: org/eclipse/paho/org.eclipse.paho.client.mqttv3/1.2.0 (maven.org)
Central Repository: org/eclipse/paho/org.eclipse.paho.android.service/1.1.1 (maven.org)
2. 将下载好的jar包加入到AndroidStudio中。（我们先新建一个project）
创建一个新项目--New Project
选择空视图
进行相关配置选择如下，点击ok即可创建一个新工程。
切换视图——Android --> Project
new一个文件命名为 lib 用来存放 jar 包。
将两个jar包复制到 lib 文件下。
右击 jar 包，选择Add As Library将包添加到配置中。
两个文件全部添加（完成后如下所示）
切换回Android视图，build.gradle可以看到自动添加的依赖，如下图。
打开manifest，添加允许
<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
在下面位置再添加一行代码。
<service android:name="org.eclipse.paho.android.service.MqttService" />
到这里配置就结束了，接下来就是mqtt及连接收发数据代码实现。
AndriodStudio代码实现
WebSocket客户端新建一个连接，修改主机名。
MQTTX建立连接
结果展示
<?xml version="1.0" encoding="utf-8"?>
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
xmlns:app="http://schemas.android.com/apk/res-auto"
xmlns:tools="http://schemas.android.com/tools"
android:layout_width="match_parent"
android:layout_height="match_parent"
android:orientation="vertical"
android:layout_gravity="center"
tools:context=".MainActivity">
<TextView
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:text="Exqm Mqtt"
android:textSize="30dp"
android:layout_gravity="center"
android:layout_margin="60dp" />
<TextView
android:id="@+id/test1"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:text="Hello World!"
android:layout_gravity="center"
android:layout_margin="60dp"
app:layout_constraintBottom_toBottomOf="parent"
app:layout_constraintEnd_toEndOf="parent"
app:layout_constraintStart_toStartOf="parent"
app:layout_constraintTop_toTopOf="parent" />
<Button
android:id="@+id/bt0"
android:text="按钮"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_gravity="center"
android:orientation="vertical" />
</LinearLayout>
package com.example.connect;
import androidx.appcompat.app.AppCompatActivity;
import android.annotation.SuppressLint;
import android.os.Bundle;
import android.os.Handler;
import android.os.Message;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.TextView;
import android.widget.Toast;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
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.persist.MemoryPersistence;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class MainActivity extends AppCompatActivity {
// 按照自己的写就行
private String host = "tcp://broker.emqx.io:1883";
private String userName = "mk";
private String passWord = "123456";
private String mqtt_id = "mqttx_a687ae08";
private String mqtt_sub_topic = "mk123";
private String mqtt_pub_topic = "mk123";
private int i = 1;
private Handler handler;
private MqttClient client;
private MqttConnectOptions options;
private ScheduledExecutorService scheduler;
private static final String TAG = "MQTT";
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
TextView text1 = findViewById(R.id.test1);
@SuppressLint({"MissingInflatedId", "LocalSuppress"})
Button btn = findViewById(R.id.bt0);
btn.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
i++;
publishMessagePlus(mqtt_pub_topic, "第一个客户端发送的信息 " + i); // Publish message
btn.setText(String.valueOf(i));
}
});
init();
startReconnect();
handler = new Handler() {
@SuppressLint({"SetTextI18n", "HandlerLeak"})
public void handleMessage(Message msg) {
super.handleMessage(msg);
switch (msg.what) {
case 1: // Example case
break;
case 2: // Example case
break;
case 3: // MQTT message received
text1.setText(msg.obj.toString());
break;
case 30: // Connection failure
Toast.makeText(MainActivity.this, "连接失败", Toast.LENGTH_SHORT).show();
break;
case 31: // Connection success
Toast.makeText(MainActivity.this, "连接成功", Toast.LENGTH_SHORT).show();
try {
Log.d(TAG, "Subscribing to topic: " + mqtt_sub_topic);
client.subscribe(mqtt_sub_topic, 1); // Subscribe with QoS 1
} catch (MqttException e) {
Log.e(TAG, "Subscription failed", e);
}
publishMessagePlus(mqtt_pub_topic, "第一个客户端发送的信息");
break;
default:
break;
}
}
};
}
private void init() {
try {
client = new MqttClient(host, mqtt_id, new MemoryPersistence());
options = new MqttConnectOptions();
options.setCleanSession(true);
options.setUserName(userName);
options.setPassword(passWord.toCharArray());
options.setConnectionTimeout(10);
options.setKeepAliveInterval(20);
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
Log.e(TAG, "Connection lost", cause);
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
Log.d(TAG, "Delivery complete: " + token.isComplete());
}
@Override
public void messageArrived(String topicName, MqttMessage message) {
Log.d(TAG, "Message arrived: " + topicName + " - " + message.toString());
Message msg = new Message();
msg.what = 3;
msg.obj = topicName + "---" + message.toString();
handler.sendMessage(msg);
}
});
} catch (MqttException e) {
Log.e(TAG, "MQTT initialization error", e);
}
}
private void Mqtt_connect() {
new Thread(new Runnable() {
@Override
public void run() {
try {
if (!client.isConnected()) {
Log.d(TAG, "Connecting to MQTT broker...");
client.connect(options);
Message msg = new Message();
msg.what = 31; // Connection success
handler.sendMessage(msg);
}
} catch (MqttException e) {
Log.e(TAG, "MQTT connection failed", e);
Message msg = new Message();
msg.what = 30; // Connection failed
handler.sendMessage(msg);
}
}
}).start();
}
private void startReconnect() {
scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
if (!client.isConnected()) {
Mqtt_connect();
}
}
}, 0 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);
}
private void publishMessagePlus(String topic, String message2) {
if (client == null || !client.isConnected()) {
Log.e(TAG, "Client is not connected. Cannot publish.");
return;
}
try {
MqttMessage message = new MqttMessage();
message.setPayload(message2.getBytes());
client.publish(topic, message);
Log.d(TAG, "Message published to topic: " + topic);
} catch (MqttException e) {
Log.e(TAG, "Error publishing message", e);
}
}
}
以上代码参考了别的大佬的教程。
三、AndroidStudio连接腾讯云
连接云只需要我们在云端配置emqx环境，然后将上面 AndroidStudio 中ip改成云服务器ip即可。
下载emqx相关配置
我们先单独建立一个包
mkdir app/soft/mqtt
在mqtt目录下下载emqx
wget https://www.emqx.com/en/downloads/broker/3.1.2/emqx-centos7-v3.1.2.zip
下载好后解压
unzip emqx-centos7-v3.1.2.zip.1
启动emqx：./bin/emqx start
修改配置文件
emqx/etc/emqx.conf
允许匿名：allow_anonymous = true
我们还是像之前一样的操作订阅主题，只不过ip为：
private String host = "tcp://云平台ip:1883";
下面的emqx客户端为云平台配置的客户端，链接为：
云平台公网ip:1808
默认登录账号：admin，密码：public。
OK，到这里一个简单的使用mqtt连接emqx进行通信就基本完成了，大家可以自行丰富内容，比如和单片机通信等。
