当前位置：首页 > news >正文

08_paho.mqtt.cpp库使用示例

news 2025/7/7 0:19:21

1.c/c++ mqtt相关库

paho MQTT C/C++ 库
mosquitto C/C++库

本文介绍 paho_mqtt_c++库使用示例

2.安装

1.下载编译

git clone https://github.com/eclipse/paho.mqtt.cpp
cd paho.mqtt.cpp
git checkout v1.4.0 # 可以按官网教程, 更新到最新版本

git submodule init
git submodule update

cmake -Bbuild -H. -DPAHO_WITH_MQTT_C=ON -DPAHO_BUILD_EXAMPLES=ON
cd build
make -j4
sudo make install

2.安装文件

c++库

libpaho-mqttpp3.so

c库

libpaho-mqtt3a.so
libpaho-mqtt3c.so

头文件

/usr/local/include/MQTTAsync.h
/usr/local/include/MQTTClient.h
/usr/local/include/MQTTClientPersistence.h
/usr/local/include/MQTTProperties.h
/usr/local/include/MQTTReasonCodes.h
/usr/local/include/MQTTSubscribeOpts.h
/usr/local/include/MQTTExportDeclarations.h
…

3.编译链接选项

由头文件位置可知, 编译选项
-I /usr/local/include

由库文件位置可以, 链接选项
-L /usr/local/lib
-lpaho-mqttpp3 -lpaho-mqtt3a -lpaho-mqtt3c

3.mqtt发布_订阅示例

1.订阅发布代码

1_mqtt_发布.cc


#include <iostream>
#include <cstdlib>
#include <string>
#include <thread> // For sleep
#include <atomic>
#include <chrono>
#include <cstring>
#include <unistd.h>
#include <stdio.h>
#include "mqtt/async_client.h"

using namespace std;

const string DFLT_SERVER_ADDRESS{"mqtt://localhost:1883"};

const string TOPIC{"test"};
const int QOS = 1;

const char *PAYLOADS[] = {
    "Hello World!",
    "Hi there!",
    "Is anyone listening?",
    "Someone is always listening.",
    nullptr};

const auto TIMEOUT = std::chrono::seconds(10);

int main(int argc, char *argv[])
{
    string address = (argc > 1) ? string(argv[1]) : DFLT_SERVER_ADDRESS;

    cout << "Initializing for server '" << address << "'..." << endl;
    mqtt::async_client cli(address, "");

    cout << "  ...OK" << endl;

    try
    {
        cout << "\nConnecting..." << endl;
        cli.connect()->wait();
        cout << "  ...OK" << endl;

        cout << "\nPublishing messages..." << endl;

        mqtt::topic top(cli, "test", QOS);
        mqtt::token_ptr tok;

        size_t i = 0;
        char buf[64];
        while (i < 1000)
        {
            snprintf(buf, sizeof(buf), "%ld %s", i, PAYLOADS[i % 4]);
            tok = top.publish(buf);
            sleep(1);
            i++;
        }
        tok->wait(); // Just wait for the last one to complete.
        cout << "OK" << endl;

        // Disconnect
        cout << "\nDisconnecting..." << endl;
        cli.disconnect()->wait();
        cout << "  ...OK" << endl;
    }
    catch (const mqtt::exception &exc)
    {
        cerr << exc << endl;
        return 1;
    }

    return 0;
}

2_mqtt_订阅.cc

#include <iostream>
#include <cstring>
#include <cstdlib>
#include <unistd.h>
#include <mqtt/async_client.h>

const std::string SERVER_ADDRESS("tcp://localhost:1883");
const std::string CLIENT_ID("paho_cpp_async");
const std::string TOPIC("test");

class callback : public virtual mqtt::callback, public virtual mqtt::iaction_listener
{
private:
    mqtt::async_client &client_;

public:
    callback(mqtt::async_client &client) : client_(client) {}

    void connection_lost(const std::string &cause) override
    {
        std::cout << "\nConnection lost" << std::endl;
        if (!cause.empty())
            std::cout << "\tcause: " << cause << std::endl;
    }

    void delivery_complete(mqtt::delivery_token_ptr token) override {}

    void message_arrived(const mqtt::const_message_ptr msg) override
    {
        std::cout << "Message arrived: " << msg->get_payload_str() << std::endl;
    }

    void on_failure(const mqtt::token &tok) override
    {
        std::cout << "Connection attempt failed" << std::endl;
    }

    void on_success(const mqtt::token &tok) override
    {
        std::cout << "Connection attempt successful" << std::endl;

        // 发布消息
        mqtt::message_ptr pubmsg = mqtt::make_message(TOPIC, "Hello from Paho C++");
        pubmsg->set_qos(1);
        client_.publish(pubmsg);
    }
};

int main(int argc, char *argv[])
{
    mqtt::async_client client(SERVER_ADDRESS, CLIENT_ID);

    callback cb(client);
    client.set_callback(cb);

    mqtt::connect_options connOpts;
    connOpts.set_clean_session(true);

    try
    {
        std::cout << "Connecting to the MQTT server..." << std::endl;
        mqtt::token_ptr conntok = client.connect(connOpts);
        conntok->wait();

        std::cout << "Connected!" << std::endl;

        // 订阅消息
        client.subscribe(TOPIC, 1);

        while (true)
        {
            // 持续监听消息
            usleep(1000000);
        }
    }
    catch (const mqtt::exception &exc)
    {
        std::cerr << "Error: " << exc.what() << std::endl;
        return 1;
    }

    return 0;
}

2.构建脚本: SConstruct

env = Environment()
env["PROGSUFFIX"] = ".out"  # 可执行文件后缀.out
env["CCFLAGS"] = " -g3 -O0 -Wall"  # gdb 调试
paho_mqtt_libs = ["paho-mqttpp3", "paho-mqtt3c", "paho-mqtt3a"]
env["LIBS"] = paho_mqtt_libs
env.Program("1_mqtt_发布.cc")
env.Program("2_mqtt_订阅.cc")

3.编译

scons
scons: Reading SConscript files …
scons: done reading SConscript files.
scons: Building targets …
g++ -o 1_mqtt_发布.o -c -g3 -O0 -Wall 1_mqtt_发布.cc
g++ -o 1_mqtt_发布.out 1_mqtt_发布.o -lpaho-mqttpp3 -lpaho-mqtt3c -lpaho-mqtt3a
g++ -o 2_mqtt_订阅.o -c -g3 -O0 -Wall 2_mqtt_订阅.cc
g++ -o 2_mqtt_订阅.out 2_mqtt_订阅.o -lpaho-mqttpp3 -lpaho-mqtt3c -lpaho-mqtt3a
scons: done building targets.

4.测试验证

终端1: ./1_mqtt_发布.out
终端2: ./2_mqtt_订阅.out
结果: 终端2 订阅端接收到的消息

./2_mqtt_订阅.out
Connecting to the MQTT server…
Connected!
Message arrived: 5 Hi there!
Message arrived: 6 Is anyone listening?
Message arrived: 7 Someone is always listening.
Message arrived: 8 Hello World!
Message arrived: 9 Hi there!
Message arrived: 10 Is anyone listening?
Message arrived: 11 Someone is always listening.
Message arrived: 12 Hello World!
Message arrived: 13 Hi there!
Message arrived: 14 Is anyone listening?
Message arrived: 15 Someone is always listening.
Message arrived: 16 Hello World!
Message arrived: 17 Hi there!