使用安全 MQTT 发布和订阅 ThingSpeak 通道
此示例展示如何使用 ESP8266 板通过安全 MQTT 在 ThingSpeak™ 中记录 RSSI 数据。您可以按照注释中所示修改代码,以使其适用于其他主板或非安全的 MQTT。
设置
1) 创建一个 ThingSpeak 通道,如在新通道中收集数据所示。
2) 通过点击 ThingSpeak 页面顶部的 Devices > MQTT,然后点击 Add Device 来创建 MQTT 设备。设置设备并将新通道添加到其授权列表时,点击 Download Credentials > Plain Text。详情请参阅创建 ThingSpeak MQTT 设备。使用下面代码部分中保存的凭据。
3) 添加 ESP8266 板包:
a.在 File > Preferences 下,在 Additional Board Manager URLs 中输入 https://arduino.esp8266.com/stable/package_esp8266com_index.json
。
b.选择 Tools > Boards > Board Manager。在搜索栏中输入 ESP8266
并安装该包。
4) 将库 WiFi.h
和 WiFiClientSecure.h
下载到您的 Arduino IDE。
代码
1) 定义无线网络、ThingSpeak MQTT 凭证以连接您的主板、连接安全性,并包含适当的库。编辑此代码以适应您的特定凭证和设置。
// Ensure correct credentials to connect to your WiFi Network. char ssid[] = "ENTER_YOUR_WIFI_SSID"; char pass[] = "ENTER_YOUR_WIFI_PASSWORD"; // Ensure that the credentials here allow you to publish and subscribe to the ThingSpeak channel. #define channelID ENTER_YOUR_THINGSPEAK_CHANNEL_NUMBER const char mqttUserName[] = "ENTER_YOUR_MQTT_USERNAME"; const char clientID[] = "ENTER_YOUR_MQTT_CLIENT_ID"; const char mqttPass[] = "ENTER_YOUR_MQTT_PASSWORD"; // It is strongly recommended to use secure connections. However, certain hardware does not work with the WiFiClientSecure library. // Comment out the following #define to use non-secure MQTT connections to ThingSpeak server. #define USESECUREMQTT // Comment the following line if not using an ESP8266. #define ESP8266BOARD #include <PubSubClient.h> #ifdef ESP8266BOARD #include <ESP8266WiFi.h> const char* PROGMEM thingspeak_cert_thumbprint = "9780c25078532fc0fd03dae01bfd8c923fff9878"; #else #include <WiFi.h> const char * PROGMEM thingspeak_ca_cert = \ "-----BEGIN CERTIFICATE-----\n" \ "MIIDxTCCAq2gAwIBAgIQAqxcJmoLQJuPC3nyrkYldzANBgkqhkiG9w0BAQUFADBs\n" \ "MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRGlnaUNlcnQgSW5jMRkwFwYDVQQLExB3\n" \ "d3cuZGlnaWNlcnQuY29tMSswKQYDVQQDEyJEaWdpQ2VydCBIaWdoIEFzc3VyYW5j\n" \ "ZSBFViBSb290IENBMB4XDTA2MTExMDAwMDAwMFoXDTMxMTExMDAwMDAwMFowbDEL\n" \ "MAkGA1UEBhMCVVMxFTATBgNVBAoTDERpZ2lDZXJ0IEluYzEZMBcGA1UECxMQd3d3\n" \ "LmRpZ2ljZXJ0LmNvbTErMCkGA1UEAxMiRGlnaUNlcnQgSGlnaCBBc3N1cmFuY2Ug\n" \ "RVYgUm9vdCBDQTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAMbM5XPm\n" \ "+9S75S0tMqbf5YE/yc0lSbZxKsPVlDRnogocsF9ppkCxxLeyj9CYpKlBWTrT3JTW\n" \ "PNt0OKRKzE0lgvdKpVMSOO7zSW1xkX5jtqumX8OkhPhPYlG++MXs2ziS4wblCJEM\n" \ "xChBVfvLWokVfnHoNb9Ncgk9vjo4UFt3MRuNs8ckRZqnrG0AFFoEt7oT61EKmEFB\n" \ "Ik5lYYeBQVCmeVyJ3hlKV9Uu5l0cUyx+mM0aBhakaHPQNAQTXKFx01p8VdteZOE3\n" \ "hzBWBOURtCmAEvF5OYiiAhF8J2a3iLd48soKqDirCmTCv2ZdlYTBoSUeh10aUAsg\n" \ "EsxBu24LUTi4S8sCAwEAAaNjMGEwDgYDVR0PAQH/BAQDAgGGMA8GA1UdEwEB/wQF\n" \ "MAMBAf8wHQYDVR0OBBYEFLE+w2kD+L9HAdSYJhoIAu9jZCvDMB8GA1UdIwQYMBaA\n" \ "FLE+w2kD+L9HAdSYJhoIAu9jZCvDMA0GCSqGSIb3DQEBBQUAA4IBAQAcGgaX3Nec\n" \ "nzyIZgYIVyHbIUf4KmeqvxgydkAQV8GK83rZEWWONfqe/EW1ntlMMUu4kehDLI6z\n" \ "eM7b41N5cdblIZQB2lWHmiRk9opmzN6cN82oNLFpmyPInngiK3BD41VHMWEZ71jF\n" \ "hS9OMPagMRYjyOfiZRYzy78aG6A9+MpeizGLYAiJLQwGXFK3xPkKmNEVX58Svnw2\n" \ "Yzi9RKR/5CYrCsSXaQ3pjOLAEFe4yHYSkVXySGnYvCoCWw9E1CAx2/S6cCZdkGCe\n" \ "vEsXCS+0yx5DaMkHJ8HSXPfqIbloEpw8nL+e/IBcm2PN7EeqJSdnoDfzAIJ9VNep\n" \ "+OkuE6N36B9K\n" \ "-----END CERTIFICATE-----\n"; #endif #ifdef USESECUREMQTT #include <WiFiClientSecure.h> #define mqttPort 8883 WiFiClientSecure client; #else #define mqttPort 1883 WiFiClient client; #endif
2) 定义连接参数并初始化 PubSubClient
实例。
const char* server = "mqtt3.thingspeak.com"; int status = WL_IDLE_STATUS; long lastPublishMillis = 0; int connectionDelay = 1; int updateInterval = 15; PubSubClient mqttClient( client );
3) 处理传入的订阅消息,订阅接收更新并发布消息。
// Function to handle messages from MQTT subscription. void mqttSubscriptionCallback( char* topic, byte* payload, unsigned int length ) { // Print the details of the message that was received to the serial monitor. Serial.print("Message arrived ["); Serial.print(topic); Serial.print("] "); for (int i = 0; i < length; i++) { Serial.print((char)payload[i]); } Serial.println(); } // Subscribe to ThingSpeak channel for updates. void mqttSubscribe( long subChannelID ){ String myTopic = "channels/"+String( subChannelID )+"/subscribe"; mqttClient.subscribe(myTopic.c_str()); } // Publish messages to a ThingSpeak channel. void mqttPublish(long pubChannelID, String message) { String topicString ="channels/" + String( pubChannelID ) + "/publish"; mqttClient.publish( topicString.c_str(), message.c_str() ); }
4) 连接到 WiFi 和 MQTT 服务器。
// Connect to WiFi. void connectWifi() { Serial.print( "Connecting to Wi-Fi..." ); // Loop until WiFi connection is successful #ifdef ESP8266BOARD while ( WiFi.waitForConnectResult() != WL_CONNECTED ) { #else while ( WiFi.status() != WL_CONNECTED ) { #endif WiFi.begin( ssid, pass ); delay( connectionDelay*1000 ); Serial.print( WiFi.status() ); } Serial.println( "Connected to Wi-Fi." ); } // Connect to MQTT server. void mqttConnect() { // Loop until connected. while ( !mqttClient.connected() ) { // Connect to the MQTT broker. if ( mqttClient.connect( clientID, mqttUserName, mqttPass ) ) { Serial.print( "MQTT to " ); Serial.print( server ); Serial.print (" at port "); Serial.print( mqttPort ); Serial.println( " successful." ); } else { Serial.print( "MQTT connection failed, rc = " ); // See https://pubsubclient.knolleary.net/api.html#state for the failure code explanation. Serial.print( mqttClient.state() ); Serial.println( " Will try again in a few seconds" ); delay( connectionDelay*1000 ); } } }
5) 初始化串行数据传输,并在 setup
方法中设置 MQTT 代理详细信息。
void setup() { Serial.begin( 115200 ); // Delay to allow serial monitor to come up. delay(3000); // Connect to Wi-Fi network. connectWifi(); // Configure the MQTT client mqttClient.setServer( server, mqttPort ); // Set the MQTT message handler function. mqttClient.setCallback( mqttSubscriptionCallback ); // Set the buffer to handle the returned JSON. NOTE: A buffer overflow of the message buffer will result in your callback not being invoked. mqttClient.setBufferSize( 2048 ); // Use secure MQTT connections if defined. #ifdef USESECUREMQTT // Handle functionality differences of WiFiClientSecure library for different boards. #ifdef ESP8266BOARD client.setFingerprint(thingspeak_cert_thumbprint); #else client.setCACert(thingspeak_ca_cert); #endif #endif }
6) 维持 MQTT 连接,并以 loop
方法定期向通道发布数据。
void loop() { // Reconnect to WiFi if it gets disconnected. if (WiFi.status() != WL_CONNECTED) { connectWifi(); } // Connect if MQTT client is not connected and resubscribe to channel updates. if (!mqttClient.connected()) { mqttConnect(); mqttSubscribe( channelID ); } // Call the loop to maintain connection to the server. mqttClient.loop(); // Update ThingSpeak channel periodically. The update results in the message to the subscriber. if ( abs(long(millis()) - lastPublishMillis) > updateInterval*1000) { mqttPublish( channelID, (String("field1=")+String(WiFi.RSSI())) ); lastPublishMillis = millis(); } }
另请参阅
Publish to a Channel Field Feed | Publish to a Channel Feed