前言

过去在製造业为了系统间沟通，我们往往是要写Tx/Rx的程式，而每次要不同的生产数据就需要额外再写另外一组传输程式，也增加了更多的维护成本，因此有需要想一个简单的机制让讯息流通更为简单，刚好IOT议题让我知道了Redis，他具有发布&订阅机制非常好用，但因为他的机制有些时候会把订阅者踢掉，不是很稳定，所以我转而去看像是RabbitMQ、EMQ X 这种讯息broker，也能启动MQTT的服务。 在此将以RabbitMQ启动MQTT服务，然后用paho-mqtt-cpp函式库来做mqtt client。

文末附上mqtt client测试code。

安装RabbitMQ

更新apt资源库：

sudo apt-get update

安装RabbitMQ服务：

sudo apt-get install rabbitmq-server

确认RabbitMQ是否启动：

sudo service rabbitmq-server status

执行完应该会看到类似以下的资讯

● rabbitmq-server.service - RabbitMQ broker   Loaded: loaded (/lib/systemd/system/rabbitmq-server.service; enabled; vendor preset: enabled)   Active: active (running) since Tue 2023-05-02 15:27:34 UTC; 3 days ago Main PID: 18509 (beam.smp)    Tasks: 220 (limit: 2295)   Memory: 325.1M      CPU: 32min 17.997s   CGroup: /system.slice/rabbitmq-server.service

启动MQTT服务

启动MQTT外挂：

sudo rabbitmq-plugins enable rabbitmq_mqtt

检查是否有启动MQTT服务：

sudo rabbitmq-plugins list

有安装的话，会在rabbitmq_mqtt那边看到E的字样代表启动mqtt

Listing plugins with pattern ".*" ... Configured: E = explicitly enabled; e = implicitly enabled | Status: * = running on rabbit@hostname |/[e*] amqp_client                        5.16.1[  ] cowboy                             1.1.2[  ] cowlib                             1.0.2[  ] rabbitmq_amqp1_0                   3.8.22[E*] rabbitmq_mqtt                      3.8.22[  ] rabbitmq_stomp                     3.8.22

安装paho-mqtt-cpp

以下安装我是在使用者的家目录下安装，可以用 cd ~ 回到家目录。 记得一定要先安装 paho-mqtt-c，再安装paho-mqtt-cpp才能顺利安装。

安装paho-mqtt-c：

sudo apt-get updatesudo apt-get install -y build-essential gcc make cmake libssl-dev gitgit clone https://github.com/eclipse/paho.mqtt.c.gitcd paho.mqtt.cmkdir build && cd buildcmake -DPAHO_ENABLE_TESTING=OFF -DPAHO_BUILD_STATIC=ON -DPAHO_WITH_SSL=ON -DPAHO_HIGH_PERFORMANCE=ON ..makesudo make install

安装paho-mqtt-cpp：

git clone https://github.com/eclipse/paho.mqtt.cpp.gitcd paho.mqtt.cppmkdir build && cd buildcmake -DPAHO_BUILD_STATIC=ON -DPAHO_BUILD_DOCUMENTATION=FALSE -DPAHO_BUILD_SAMPLES=FALSE -DPAHO_WITH_SSL=ON ..makesudo make install

configure dynamic linker

sudo ldconfig

mqtt client cpp 範例

编译範例code的shell指令

g++ -std=c++11 -o mqtt_example mqtt_example.cpp -lpaho-mqttpp3 -lpaho-mqtt3a

其中档案名称为mqtt_example.cpp
-o 代表输出的档名
-l 代表动态连结
paho-mqttpp3 是c++版本函式库
paho-mqtt3a 是c版本函式库，若mqtt有设定帐密，则可用paho-mqtt3as

範例code

#include <iostream>#include <cstring>#include "mqtt/async_client.h"const std::string ADDRESS { "tcp://localhost:1883" };const std::string CLIENT_ID { "paho_cpp_example" };const std::string TOPIC { "test/topic" };class callback : public virtual mqtt::callback {public:    virtual void connection_lost(const std::string& cause) override {        std::cout << "Connection lost: " << cause << std::endl;    }    virtual void delivery_complete(mqtt::delivery_token_ptr tok) override {        std::cout << "Delivery complete" << std::endl;    }    virtual void message_arrived(mqtt::const_message_ptr msg) override {        std::cout << "Message received: " << msg->to_string() << std::endl;    }};int main(int argc, char* argv[]) {    mqtt::async_client client(ADDRESS, CLIENT_ID);    callback cb;    client.set_callback(cb);    mqtt::connect_options connOpts;    connOpts.set_keep_alive_interval(20);    connOpts.set_clean_session(true);    try {        client.connect(connOpts)->wait_for(std::chrono::seconds(5));    }    catch (const mqtt::exception& exc) {        std::cerr << "Error: " << exc.what() << std::endl;        return 1;    }    // Subscribe to the topic    client.subscribe(TOPIC, 1)->wait_for(std::chrono::seconds(5));    // Publish a message to the topic    std::string payload { "Hello, world!" };    mqtt::message_ptr pubmsg = mqtt::make_message(TOPIC, payload);    pubmsg->set_qos(1);    client.publish(pubmsg)->wait_for(std::chrono::seconds(5));    // Wait for messages to arrive    std::this_thread::sleep_for(std::chrono::seconds(5));    client.disconnect()->wait_for(std::chrono::seconds(5));    return 0;}