Android、ESP32、ESP8266的mqtt通信

Android

activity_main

<?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"tools:context=".MainActivity"android:orientation="vertical"><TextViewandroid:id="@+id/recvTxt"android:layout_width="match_parent"android:layout_height="600dp"android:text=""app:layout_constraintBottom_toBottomOf="parent"app:layout_constraintEnd_toEndOf="parent"app:layout_constraintStart_toStartOf="parent"app:layout_constraintTop_toTopOf="parent" /><Buttonandroid:id="@+id/onBtn"android:layout_width="match_parent"android:layout_height="wrap_content"android:text="开灯" /><Buttonandroid:id="@+id/offBtn"android:layout_width="match_parent"android:layout_height="wrap_content"android:text="关灯" /></LinearLayout>

AndroidManifest

<uses-permission android:name="android.permission.INTERNET"/>
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE"/>

MainActivity

package com.example.myapplication;
/*
* https://repo.eclipse.org/content/repositories/paho-releases/org/eclipse/paho/
* https://repo.eclipse.org/content/repositories/paho-releases/org/eclipse/paho/org.eclipse.paho.client.mqttv3/1.2.5/
* 下载org.eclipse.paho.client.mqttv3-1.2.5.jar文件
* 将下载的jar包复制至项目libs目录下，并右击mqtt jar包 ADD As Libray.. ，将mqtt jar包导入库文件中。
* */
import androidx.annotation.NonNull;
import androidx.appcompat.app.AppCompatActivity;import android.os.Bundle;
import android.os.Handler;
import android.os.Message;
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;public class MainActivity extends AppCompatActivity {private String host = "tcp://broker.emqx.io:1883";private String userName = "dsx_phone";private String passWord = "dsx_phone_pwd";private String mqtt_id ="dsx_phone_001";private MqttClient client;private String subTopic = "dsx_phone"; //订阅的主题  接收内容private String sendTopic = "dsx_esp8266";//发送的主题private MqttConnectOptions options;private TextView recvTxt;private Button onBtn,offBtn;MyHandler myHandler;@Overrideprotected void onCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);setContentView(R.layout.activity_main);recvTxt = findViewById(R.id.recvTxt);onBtn = findViewById(R.id.onBtn);offBtn = findViewById(R.id.offBtn);myHandler = new MyHandler();connectServer();onBtn.setOnClickListener(new View.OnClickListener() {@Overridepublic void onClick(View view) {publishMsg(sendTopic,"on");}});offBtn.setOnClickListener(new View.OnClickListener() {@Overridepublic void onClick(View view) {publishMsg(sendTopic,"off");}});}private void connectServer(){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() {@Overridepublic void connectionLost(Throwable throwable) {//重新连接}@Overridepublic void messageArrived(String s, MqttMessage mqttMessage) throws Exception {//收到消息Message msg = new Message();msg.what = 2;msg.obj = s + "------" + mqttMessage.toString();myHandler.sendMessage(msg);}@Overridepublic void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {//发送完成后执行到这里Message msg = new Message();msg.what = 4;msg.obj = "Publish OK".toString();myHandler.sendMessage(msg);}});new Thread(new Runnable() {@Overridepublic void run() {try {if(!(client.isConnected())){client.connect(options);Message msg = new Message();msg.what = 1;msg.obj = "Connect Server Success".toString();myHandler.sendMessage(msg);publishMsg(sendTopic,"我是来自手机端的消息!");}}catch (Exception e){e.printStackTrace();Message msg = new Message();msg.what = 3;msg.obj = "Connect Server Fail".toString();myHandler.sendMessage(msg);}}}).start();} catch (MqttException e) {e.printStackTrace();throw new RuntimeException(e);}}private void publishMsg(String send_topic,String msg){if(client == null || !(client.isConnected())){return;}MqttMessage mqtt_msg = new MqttMessage();mqtt_msg.setPayload(msg.getBytes());try {client.publish(send_topic,mqtt_msg);} catch (MqttException e) {e.printStackTrace();throw new RuntimeException(e);}}class MyHandler extends Handler{@Overridepublic void handleMessage(@NonNull Message msg) {super.handleMessage(msg);switch (msg.what){case 1:recvTxt.setText(msg.obj.toString());Toast.makeText(MainActivity.this,msg.obj.toString(),Toast.LENGTH_LONG).show();try {client.subscribe(subTopic);} catch (MqttException e) {e.printStackTrace();throw new RuntimeException(e);}break;case 2:recvTxt.setText(msg.obj.toString());Toast.makeText(MainActivity.this,"Recv Data",Toast.LENGTH_LONG).show();break;case 3:recvTxt.setText(msg.obj.toString());Toast.makeText(MainActivity.this,msg.obj.toString(),Toast.LENGTH_LONG).show();break;case 4:recvTxt.setText(msg.obj.toString());Toast.makeText(MainActivity.this,msg.obj.toString(),Toast.LENGTH_LONG).show();break;case 5:Toast.makeText(MainActivity.this,"5",Toast.LENGTH_LONG).show();break;default:Toast.makeText(MainActivity.this,"default",Toast.LENGTH_LONG).show();break;}}}
}

ESP32

/*
//wifi连接
#include <WiFi.h>
const char* id="TP-LINK_8F3882";   //定义两个字符串指针常量
const char* psw="123456789";
void setup() {Serial.begin(115200);WiFi.begin(id,psw);while(WiFi.status() != WL_CONNECTED){			//未连接上delay(500);Serial.println("wifi connecting...");}Serial.println("wifi connect ok!");				//连接上
}
void loop(){							//空循环}*/
/*Broker:
broker.emqx.io
TCP 端口:
1883
WebSocket 端口:
8083
SSL/TLS 端口:
8883
WebSocket Secure 端口:
8084
QUIC 端口:
14567
CA 证书文件:
broker.emqx.io-ca.crt
*/#include <WiFi.h>
#include <PubSubClient.h>// const char* ssid     = "Netcore_dsx";
// const char* password = "dsx54254";const char* ssid="TP-LINK_8F3882";   //定义两个字符串指针常量
const char* password="123456789";//led
#define JDQ 2const char* MQTT_SERVER  = "broker.emqx.io";
const int   MQTT_PORT    = 1883;
const char* MQTT_USRNAME = "admin";
const char* MQTT_PASSWD  = "adminadmin";
const char* TOPIC = "dsx_esp8266";     //订阅的主题  接收这个主题的消息
const char* CLIENT_ID    = "dsx_esp32_01";  //当前设备的clientid标志WiFiClient espClient;
PubSubClient  client(espClient);String send_msg="I am esp32";
String send_topic = "dsx_phone";   //往这个主题发送消息
void setup()
{Serial.begin(115200);delay(5000);Serial.println();Serial.println();Serial.print("Connecting to ");Serial.println(ssid);WiFi.begin(ssid, password);while (WiFi.status() != WL_CONNECTED) {delay(2500);Serial.print(".");}Serial.println("");Serial.println("WiFi connected");Serial.print("IP address: ");Serial.println(WiFi.localIP());pinMode(JDQ, OUTPUT);client.setServer(MQTT_SERVER, MQTT_PORT); //设定MQTT服务器与使用的端口，1883是默认的MQTT端口client.setCallback(callback);        //设定回调方式，当ESP8266收到订阅消息时会调用此方法
}void reconnect() {while (!client.connected()) {Serial.println("Attempting MQTT connection...");if (client.connect(CLIENT_ID,MQTT_USRNAME,MQTT_PASSWD)) {Serial.println("Mqtt server connected success");// 连接成功时订阅主题client.subscribe(TOPIC);Serial.println("Topic  dsx_esp32  subscribe success");pubmsg(send_topic,send_msg);} else {Serial.print("failed, rc=");Serial.print(client.state());Serial.println(" try again in 5 seconds");delay(5000);}}
}void callback(char* topic, byte* payload, unsigned int length) {Serial.print("Message arrived in topic: ");Serial.println(topic);Serial.print("Message: ");String message;for (int i = 0; i < length; i++) {message += (char) payload[i];  // Convert *byte to string}Serial.print(message);if (message == "on") {digitalWrite(JDQ, HIGH);  // Turn on the LEDpubmsg(send_topic, "LED ON");}if (message == "off") {digitalWrite(JDQ, LOW); // Turn off the LEDpubmsg(send_topic, "LED OFF");}Serial.println();Serial.println("-----------------------");
}// topicString   参数类似  "device/date" 
// messageString 参数类似  String realmsg="";void pubmsg( const String &topicString, const  String &messageString){char publishTopic[topicString.length() + 1];  strcpy(publishTopic, topicString.c_str());char publishMsg[messageString.length() + 1];   strcpy(publishMsg, messageString.c_str());// 实现ESP8266向主题发布信息if(client.publish(publishTopic, publishMsg)){Serial.print("Publish Topic:");Serial.println(publishTopic);Serial.print("Publish message:");Serial.println(publishMsg);   Serial.println("Send Success"); } else {Serial.println("Message Publish Failed."); }
}void loop()
{if (!client.connected()) {reconnect();}client.loop();
}

ESP8266

#include <ESP8266WiFi.h>
#include <PubSubClient.h>// 开发板上LED GPIO2 对应 D4  可以使用D4  也可以使用2    低电平点亮
#define LED 2// WiFi
const char *ssid = "Netcore_dsx"; // Enter your WiFi name
const char *password = "dsx54254";  // Enter WiFi password// MQTT Broker
const char *mqtt_broker = "broker.emqx.io";
const char *topic = "dsx_esp8266";
const char *mqtt_username = "emqx_dsx_esp8266";
const char *mqtt_password = "public_dsx_esp8266";
const int mqtt_port = 1883;
const char *send_topic = "dsx_phone";
bool ledState = false;WiFiClient espClient;
PubSubClient client(espClient);void setup() {// Set software serial baud to 115200;Serial.begin(115200);delay(1000); // Delay for stability// Connecting to a WiFi networkWiFi.begin(ssid, password);while (WiFi.status() != WL_CONNECTED) {delay(500);Serial.println("Connecting to WiFi...");}Serial.println("Connected to the WiFi network");// Setting LED pin as outputpinMode(LED, OUTPUT);digitalWrite(LED, HIGH);  // Turn off the LED initially// Connecting to an MQTT brokerclient.setServer(mqtt_broker, mqtt_port);client.setCallback(callback);while (!client.connected()) {String client_id = "esp8266-client-";client_id += String(WiFi.macAddress());Serial.printf("The client %s connects to the public MQTT broker\n", client_id.c_str());if (client.connect(client_id.c_str(), mqtt_username, mqtt_password)) {Serial.println("Public EMQX MQTT broker connected");} else {Serial.print("Failed with state ");Serial.print(client.state());delay(2000);}}// Publish and subscribeclient.publish(send_topic, "hello phone");client.subscribe(topic);
}void callback(char *topic, byte *payload, unsigned int length) {Serial.print("Message arrived in topic: ");Serial.println(topic);Serial.print("Message: ");String message;for (int i = 0; i < length; i++) {message += (char) payload[i];  // Convert *byte to string}Serial.print(message);if (message == "on" && !ledState) {digitalWrite(LED, LOW);  // Turn on the LEDledState = true;client.publish(send_topic, "LED ON");}if (message == "off" && ledState) {digitalWrite(LED, HIGH); // Turn off the LEDledState = false;client.publish(send_topic, "LED OFF");}Serial.println();Serial.println("-----------------------");
}void loop() {client.loop();delay(100); // Delay for a short period in each loop iteration
}