前文安装完MQTT服务器程序后，通过命令行发出或接收数据（pub或sub），用于测试服务器，仅仅是构建系统的第一步。下面更重要的一步是客户端程序，如嵌入式系统ESP8266，通过程序访问MQTT服务器，通过程序与之进行数据交换。

老规矩，一切从例题开始。下面这段程序的最原始代码来自于在arduino IDE中安装PubSubClient.h后，其自带的例题。位于\Arduino\libraries\PubSubClient\examples\mqtt_esp8266文件夹中。
这段ESP8266的程序我把它编译长传到ModeMCU中，可以成功发送某一主题下的信息，同时通过回调函数，接收令一主题下的信息。
程序中，我对两个蚊子公司的官方mqtt服务器进行测试，但好像速度很慢，测试时大量数据丢失（我用命令行测试pub没有问题）。同时也测试了我自己安装在树莓派4上的mqtt服务器，效果很好。
在NodeMCU上的程序运行之前，需要先在树莓派上执行订阅命令：
 $ mosquitto_sub -h test.mosquitto.org -t "my_outTopic" -v
订阅服务器 test.mosquitto.org上主题为 "my_outTopic"的信息。

如果是通过树莓派上的mqtt服务器收发数据，则命令行中服务器的地址可以省略:
 $ mosquitto_sub -t "my_outTopic" -v
于是就可以在树莓派的终端窗口中看到NodeMCU每2秒钟发出的数据。

同时可以再打开一个终端串口，执行
 $ mosquitto_pub -h test.mosquitto.org -t "my_inTopic" -m "Hello tiger."
就可以在arduino的串口窗口中看到它订阅的信息。

到此，全部测试完成。

/*
 Basic ESP8266 MQTT example
 This sketch demonstrates the capabilities of the pubsub library in combination
 with the ESP8266 board/library.
 这里实现了ESP8266和MQTT服务器之间的双向数传。
 It connects to an MQTT server then:
  - publishes "hello world" to the topic "my_outTopic" every two seconds
  - subscribes to the topic "my_inTopic", printing out any messages
    it receives. NB - it assumes the received payloads are strings not binary
  - If the first character of the topic "inTopic" is an 1, switch ON the ESP Led,
    else switch it off

 It will reconnect to the server if the connection is lost using a blocking
 reconnect function. See the 'mqtt_reconnect_nonblocking' example for how to
 achieve the same result without blocking the main loop.

 To install the ESP8266 board, (using Arduino 1.6.4+):
  - Add the following 3rd party board manager under "File -> Preferences -> Additional Boards Manager URLs":
       http://arduino.esp8266.com/stable/package_esp8266com_index.json
  - Open the "Tools -> Board -> Board Manager" and click install for the ESP8266"
  - Select your ESP8266 in "Tools -> Board"
*/

#include <ESP8266WiFi.h>
#include <PubSubClient.h>

// Update these with values suitable for your network.

const char* ssid = "Xiaomi_C888";
const char* password = "xxxxxxxx";
//const char* mqtt_server = "192.168.31.10";  // 自建的树莓派MQTT服务器。OK
//const char* mqtt_server = "test.mosquitto.org"; //不成功，但用命令行可以。
const char* mqtt_server = "broker.mqtt-dashboard.com"; //另一个官方服务器

WiFiClient espClient;
PubSubClient client(espClient);
long lastMsg = 0;
char msg[50];
int value = 0;
void setup_wifi();
void callback(char* topic, byte* payload, unsigned int length);

//-------------------------------------------------------------------------------
void setup() {
  pinMode(BUILTIN_LED, OUTPUT);     // Initialize the BUILTIN_LED pin as an output
  Serial.begin(115200);
  setup_wifi();
  client.setServer(mqtt_server, 1883);
  client.setCallback(callback);
}
//-------------------------------------------------------------------------------
void setup_wifi() {
  delay(10);
  Serial.println();
  Serial.print("Connecting to ");
  Serial.println(ssid);
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }
  Serial.println("");
  Serial.println("WiFi connected");
  Serial.println("IP address: ");
  Serial.println(WiFi.localIP());
}

//------------------------------------------------------------------------------
//回调函数，只要有数据来，马上执行之：

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();

  // Switch on the LED if an 1 was received as first character
  if ((char)payload[0] == '1') {
    digitalWrite(BUILTIN_LED, LOW);   // Turn the LED on (Note that LOW is the voltage level
    // but actually the LED is on; this is because
    // it is acive low on the ESP-01)
  } else {
    digitalWrite(BUILTIN_LED, HIGH);  // Turn the LED off by making the voltage HIGH
  }

}

//--------------------------------------------------------------------------------
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("my_outTopic", "hello world");
      // ... and resubscribe
      client.subscribe("my_inTopic");
    } 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();

  long now = millis();
  if (now - lastMsg > 2000) {
    lastMsg = now;
    ++value;
    snprintf (msg, 50,"hello Tiger #%ld", value);
    Serial.print("Publish message: ");
    Serial.println(msg);
    client.publish("my_outTopic", msg);
  }
}