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

C++的MQTT开发：使用Paho的C++接口实现消息发布、订阅、连接RabbitMQ

news 文章来源：https://blog.csdn.net/goldtimes/article/details/139383700 2024/11/20 4:23:01

C++ Paho实现MQTT消息发布功能

要使用paho的cpp接口实现发布MQTT消息的功能，需要进行以下步骤：

安装paho库：首先从paho官方网站下载并安装paho的C++库。可以从https://www.eclipse.org/paho/clients/cpp/ 下载适合操作系统的版本。
创建MQTT客户端：可以使用mqtt::client类来创建一个客户端，如下所示：

mqtt::client client("tcp://broker.example.com:1883", "clientId");

在上面的代码中，broker.example.com是您的MQTT代理服务器的地址，1883是MQTT代理服务器的默认端口。clientId是客户端的唯一标识符，可以自己选择一个适合的名字。

设置连接选项：创建客户端后可以设置一些连接选项，例如设置用户名和密码，设置遗嘱消息等。以下是示例代码：

mqtt::connect_options connOpts;
connOpts.set_user_name("username");
connOpts.set_password("password");
connOpts.set_will(mqtt::message("topic", "offline", 1, true));

在上面的代码中，username和password是您的MQTT代理服务器的登录凭据。topic是遗嘱消息的主题，offline是遗嘱消息的内容，1是遗嘱消息的QoS级别（Quality of Service），true表示遗嘱消息是保留的。

连接到MQTT代理服务器：使用mqtt::client对象的connect方法连接到MQTT代理服务器，如下所示：

client.connect(connOpts);

发布消息：使用mqtt::client对象的publish方法发布消息。以下是一个示例代码：

std::string payload = "Hello, MQTT!";
client.publish("topic", payload.c_str(), payload.length());

在上面的代码中，topic是消息的主题，payload是消息的内容。您可以根据需要修改这些值。

断开连接：在完成消息发布后，您可以使用mqtt::client对象的disconnect方法断开与MQTT代理服务器的连接，如下所示：

client.disconnect();

这是使用paho的cpp接口发布MQTT消息的基本步骤，实际应用中可能需要处理更多的错误和异常情况。参考paho的官方文档和示例代码来进一步了解和掌握paho的cpp接口的使用。

完整的C++ Paho消息发布的代码演示

#include <iostream>
#include <cstring>
#include "mqtt/async_client.h"const std::string SERVER_ADDRESS("tcp://broker.example.com:1883");
const std::string CLIENT_ID("clientId");
const std::string TOPIC("topic");class mqtt_callback : public virtual mqtt::callback
{void connection_lost(const std::string& cause) override{std::cout << "\nConnection lost! Cause: " << cause << std::endl;}void delivery_complete(mqtt::delivery_token_ptr token) override{std::cout << "Message delivery complete!" << std::endl;}
};int main(int argc, char* argv[])
{try{mqtt::async_client client(SERVER_ADDRESS, CLIENT_ID);mqtt::connect_options conn_opts;mqtt_callback cb;client.set_callback(cb);client.connect(conn_opts);std::string payload = "Hello, MQTT!";mqtt::message_ptr pubmsg = mqtt::make_message(TOPIC, payload);client.publish(pubmsg);client.disconnect();}catch (const mqtt::exception& exc){std::cerr << "Error: " << exc.what() << std::endl;return 1;}return 0;
}

在这段代码中，我们包含了必要的头文件，并定义了MQTT服务器地址、客户端ID和主题。我们还定义了一个自定义的回调类mqtt_callback，它继承自mqtt::callback，用于处理连接丢失和消息传递完成事件。

在main()函数内部创建了一个带有服务器地址和客户端ID的mqtt::async_client对象。我们还创建了一个mqtt::connect_options对象来指定连接选项。使用client.set_callback(cb)为客户端设置回调函数，其中cb是mqtt_callback类的一个实例。

我们使用client.connect(conn_opts)建立与MQTT服务器的连接。如果连接成功，我们使用mqtt::make_message()创建一个带有所需负载和主题的MQTT消息。

最后，使用client.publish(pubmsg)发布消息，并使用client.disconnect()断开与MQTT服务器的连接。

客户端：C++ Paho实现 MQTT的客户端

同步客户端：mqtt::client

在paho库中，mqtt::client和mqtt::async_client是MQTT客户端的两种不同实现方式。

mqtt::client是同步的客户端实现，它在发送和接收消息时会阻塞当前线程，直到操作完成或发生超时。这意味着您需要手动管理线程和处理并发操作。使用mqtt::client可以更容易地编写简单的同步代码，特别适用于简单的MQTT应用程序或在没有并发需求的情况下。

以下是使用mqtt::client的示例代码：

mqtt::client client("tcp://broker.example.com:1883", "clientId");// Connect to the MQTT broker
client.connect();// Publish a message
client.publish("topic", "Hello, MQTT!");// Subscribe to a topic
client.subscribe("topic");// Wait for incoming messages
mqtt::message_ptr msg = client.consume_message();// Disconnect from the MQTT broker
client.disconnect();

异步客户端：mqtt::async_client

另一方面，mqtt::async_client是异步的客户端实现，它使用了异步操作和回调函数来处理发送和接收消息，不会阻塞当前线程。这意味着您可以在一个线程中同时处理多个操作，包括异步地发送和接收消息。使用mqtt::async_client可以更好地处理并发操作和复杂的MQTT应用程序。

以下是使用mqtt::async_client的示例代码：

mqtt::async_client client("tcp://broker.example.com:1883", "clientId");// Connect to the MQTT broker
mqtt::token_ptr conntok = client.connect();
conntok->wait();// Publish a message
mqtt::token_ptr pubtok = client.publish("topic", "Hello, MQTT!");
pubtok->wait();// Subscribe to a topic
mqtt::token_ptr subtok = client.subscribe("topic");
subtok->wait();// Wait for incoming messages
client.start_consuming();// Disconnect from the MQTT broker
mqtt::token_ptr disctok = client.disconnect();
disctok->wait();

虽然mqtt::async_client提供了更强大的功能和更好的并发性能，但它需要更多的代码和处理异步回调函数。如果应用程序需要处理大量的并发操作或实现复杂的MQTT逻辑，推荐使用mqtt::async_client。但如果只需要简单的同步操作或在单线程环境中运行较简单的MQTT应用程序，那么使用mqtt::client可能更合适。应该根据您的具体需求和应用程序的复杂性选择适合的客户端实现方式。

使用C++ Paho连接rabbitmq

下面是一个简单的示例代码，演示了如何使用Paho C++库连接到RabbitMQ并发布和订阅消息：

#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_topic");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;
}

这个示例代码创建了一个异步的MQTT客户端，并连接到RabbitMQ服务器。它使用一个回调类来处理与连接、消息到达等事件相关的回调。在on_success回调中，它发布一条消息。在message_arrived回调中，它处理接收到的消息。

编译和运行代码：使用适当的编译器和选项，将代码编译为可执行文件。确保已经安装了Paho C++库，并在编译时链接到该库。然后运行可执行文件。

这样就可以使用Paho C++库连接到RabbitMQ并进行消息发布和订阅操作了。

本示例代码仅提供了一个基本的框架，需要根据自己的需求进行更多的自定义和错误处理。此外，确保已正确配置RabbitMQ服务器的连接参数，如服务器地址、端口和认证信息等。

补充说明

连续发送消息的注意事项

在连续发送多条消息时避免频繁断开和重连MQTT客户端。频繁的断开和重连会增加网络开销和延迟，并且可能会对MQTT服务器造成不必要的负担。保持MQTT客户端的持久连接，并在需要时使用同一连接发送多条消息。

以下示例展示如何保持MQTT客户端的持久连接，并连续发送多条消息：

mqtt::async_client client("tcp://broker.example.com:1883", "clientId");// Connect to the MQTT broker
mqtt::connect_options connOpts;
connOpts.set_keep_alive_interval(60);  // 设置心跳间隔为60秒
connOpts.set_clean_session(false);     // 设置为非清理会话模式
mqtt::token_ptr conntok = client.connect(connOpts);
conntok->wait();// Publish multiple messages
for (int i = 0; i < 10; ++i) {std::string topic = "topic";std::string payload = "Message " + std::to_string(i);mqtt::message_ptr pubmsg = mqtt::make_message(topic, payload);mqtt::token_ptr pubtok = client.publish(pubmsg);pubtok->wait();
}// Disconnect from the MQTT broker
mqtt::token_ptr disctok = client.disconnect();
disctok->wait();

在上述示例中，使用set_keep_alive_interval()方法设置了心跳间隔为60秒，这样可以确保MQTT客户端与服务器保持持久连接。我们还使用set_clean_session(false)将客户端设置为非清理会话模式，这样即使客户端断开连接，会话状态也会保留。

然后，我们使用一个循环来连续发布多条消息。每次循环中，我们创建一个新的消息并使用client.publish()发送。通过保持持久连接，我们可以在同一连接上连续发送多条消息，而不需要频繁断开和重连。

最后，使用client.disconnect()来断开与MQTT服务器的连接。

具体的MQTT应用程序可能会有不同的需求和限制。如果应用程序需要在发送消息之间有较长的时间间隔，或者需要处理长时间的非活动状态，那么可以考虑在一段时间后断开连接，并在需要时重新连接。为了连续发送多条消息，建议保持MQTT客户端的持久连接，避免频繁断开和重连。这样可以减少网络开销并提高性能。

如果无需同步，是否需要wait

如果不需要等待异步操作完成，那么在使用mqtt::token对象的时候可以选择不调用wait方法。wait方法会阻塞当前线程，直到操作完成或发生超时。如果您不调用wait方法，程序将继续执行下一行代码，而不会等待操作完成。

以下是一个不调用wait方法的情况：

mqtt::async_client client("tcp://broker.example.com:1883", "clientId");// Connect to the MQTT broker
mqtt::token_ptr conntok = client.connect();
// 不调用wait方法，程序继续执行下一行代码// Publish a message
mqtt::token_ptr pubtok = client.publish("topic", "Hello, MQTT!");
// 不调用wait方法，程序继续执行下一行代码// Subscribe to a topic
mqtt::token_ptr subtok = client.subscribe("topic");
// 不调用wait方法，程序继续执行下一行代码// Wait for incoming messages
client.start_consuming();// Disconnect from the MQTT broker
mqtt::token_ptr disctok = client.disconnect();
// 不调用wait方法，程序继续执行下一行代码

上述示例创建了一个mqtt::async_client对象，并进行了连接、发布消息、订阅主题和断开连接的操作。但是我们没有调用wait方法等待操作完成。这意味着程序将继续执行下一行代码，而不会等待操作完成。

请注意，如果不调用wait方法，将无法确定操作是否成功完成。如果需要获取操作的结果或处理操作的回调函数，可能需要调用wait_for_completion方法或使用其他适当的方式来处理异步操作的结果。