MQTT esp8266 client.subscribe() 不工作

Question

我在一个基于物联网的项目中工作，该项目希望将我的数据存储到数据库中，并使用 mqtt 在客户端和 esp8266 之间进行通信。我尝试在 esp8266 节点 mcu 中同时实现 mysql 和 mqtt。在一个循环中，我首先检查 mqtt 消息是否到达，然后使用传感器值更新数据库。 Client.publish() 工作，但 Client.suscribe() 在数据库更新完成时不起作用。但是当只有 mqtt 完成时它工作正常。

#include <ESP8266WiFi.h>
#include <WiFiClient.h>
#include <MySQL_Connection.h>
#include <MySQL_Cursor.h>
#include <PubSubClient.h>
IPAddress server_addr(***, , ,); // IP of the MySQL server
char user[] = "root"; // MySQL user login username
char password[] = ""; // MySQL user login password
char ssid[] = "***"; // your SSID
char pass[] = "*****"; // your SSID Password
const char mqtt_server = "192.168.0.109";

long lastMsg = 0;
char msg[50];
int value = 0;
WiFiClient espClient;
MySQL_Connection conn((Client *)&espClient);

PubSubClient client(espClient);

void setup() {
Serial.begin(115200);
WiFi.begin(ssid, pass); // initializing the WIFI library

while ( WiFi.status() != WL_CONNECTED ) { // while loop to write dots     during connecting
delay ( 500 );
Serial.print ( "." );
}

// print out information about the WIFI connection
Serial.println ( "" );
Serial.print ( "Connected to " );

Serial.println ( ssid );
Serial.print ( "IP address: " );
Serial.println ( WiFi.localIP() );

client.setServer(mqtt_server, 1883);
client.setCallback(callback);
}

void callback(char* topic, byte* payload, unsigned int length) {
Serial.print("Message arrived [");
Serial.print(topic);
Serial.print("] ");
for (int i = 0; i < length; i++) {
Serial.print((char)payload[i]);
}
Serial.println();

}

void reconnect() {
// Loop until we're reconnected
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// Attempt to connect
if (client.connect("ESP8266Client")) {
Serial.println("connected");
// Once connected, publish an announcement...
client.publish("outTopic", "SAAIL");
// ... and resubscribe

  client.subscribe("say");

} else {
  Serial.print("failed, rc=");
  Serial.print(client.state());
  Serial.println(" try again in 5 seconds");
  // Wait 5 seconds before retrying
  delay(5000);
}

}
} 


void loop() {

if (!client.connected()) {
reconnect();
}
client.loop();
delay(1000);
long now = millis();
if (now - lastMsg > 2000) {
lastMsg = now;

int newTemp = sht1x.readTemperatureC();
int newHum = sht1x.readHumidity();
Serial.print("temp:");
Serial.print(newTemp);
char INSERT_SQL[] = "INSERT INTO test.users (humidity,temp) VALUES (%d, %d );";
char query[255];
sprintf(query, INSERT_SQL, newHum, newTemp);
Serial.println("Recording data.");
conn.connect(server_addr, 3306, user, password);
// Initiate the query class instance
MySQL_Cursor *cur_mem = new MySQL_Cursor(&conn);
// Execute the query
cur_mem->execute(query);
// Note: since there are no results, we do not need to read any data
// Deleting the cursor also frees up memory used
delete cur_mem;
conn.close();

}

}

Answer 1

实际上，您的代码仅在循环内不断移动，并且您正在调用客户端。 subscribe() 在 reconnect() 函数中。因此，如果您的 ESP8266 一次性连接到 MQTT Broker，那么您的 Reconnect() 将不会被调用。

这里是代码

#include <ESP8266WiFi.h>
#include <WiFiClient.h>
#include <MySQL_Connection.h>
#include <MySQL_Cursor.h>
#include <PubSubClient.h>
IPAddress server_addr(***, , ,); // IP of the MySQL server
char user[] = "root"; // MySQL user login username
char password[] = ""; // MySQL user login password
char ssid[] = "***"; // your SSID
char pass[] = "*****"; // your SSID Password
const char mqtt_server = "192.168.0.109";

long lastMsg = 0;
char msg[50];
int value = 0;
WiFiClient espClient;
MySQL_Connection conn((Client *)&espClient);

PubSubClient client(espClient);

void setup() {
Serial.begin(115200);
WiFi.begin(ssid, pass); // initializing the WIFI library

while ( WiFi.status() != WL_CONNECTED ) { // while loop to write dots     during connecting
delay ( 500 );
Serial.print ( "." );
}

// print out information about the WIFI connection
Serial.println ( "" );
Serial.print ( "Connected to " );

Serial.println ( ssid );
Serial.print ( "IP address: " );
Serial.println ( WiFi.localIP() );

client.setServer(mqtt_server, 1883);
client.setCallback(callback);
connectmqtt();  
}

void callback(char* topic, byte* payload, unsigned int length) {
Serial.print("Message arrived [");
Serial.print(topic);
Serial.print("] ");
for (int i = 0; i < length; i++) {
Serial.print((char)payload[i]);
}
Serial.println();

}

void reconnect() {
// Loop until we're reconnected
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// Attempt to connect
if (client.connect("ESP8266Client")) {
Serial.println("connected");
// Once connected, publish an announcement...
client.publish("outTopic", "SAAIL");
// ... and resubscribe

  client.subscribe("say");

} else {
  Serial.print("failed, rc=");
  Serial.print(client.state());
  Serial.println(" try again in 5 seconds");
  // Wait 5 seconds before retrying
  delay(5000);
}

}
} 


void loop() {

if (!client.connected()) {
reconnect();
}
client.loop();
delay(1000);
long now = millis();
if (now - lastMsg > 2000) {
lastMsg = now;

int newTemp = sht1x.readTemperatureC();
int newHum = sht1x.readHumidity();
Serial.print("temp:");
Serial.print(newTemp);
char INSERT_SQL[] = "INSERT INTO test.users (humidity,temp) VALUES (%d, %d );";
char query[255];
sprintf(query, INSERT_SQL, newHum, newTemp);
Serial.println("Recording data.");
conn.connect(server_addr, 3306, user, password);
// Initiate the query class instance
MySQL_Cursor *cur_mem = new MySQL_Cursor(&conn);
// Execute the query
cur_mem->execute(query);
// Note: since there are no results, we do not need to read any data
// Deleting the cursor also frees up memory used
delete cur_mem;
conn.close();

}

}

void connectmqtt()
{
  client.connect("ESP8266Client");
  {
      Serial.println("connected");
      // Once connected, publish an announcement...

      // ... and resubscribe
      client.subscribe("say");
       client.publish("outTopic", "SAAIL");

       if (!client.connected()) 
  {
    reconnect();
  }
}

}