MQTT mit C++: Beispiele mit Paho C++ und async_mqtt

Der schnellste Weg zu MQTT mit C++ ist der Eclipse-Paho-C++-Client (paho.mqtt.cpp): Du erzeugst einen mqtt::async_client, verbindest dich mit einem connect_options_builder und rufst dann publish() und subscribe() auf – jeder Aufruf liefert ein Token, auf das du mit wait() warten kannst. Wenn deine Anwendung schon auf Boost.Asio läuft, ist async_mqtt eine Header-only-Alternative, die sich nahtlos in dieses Modell einfügt.

Dieses Tutorial baut mit Paho C++ einen Publisher und einen Subscriber, ergänzt TLS, Reconnects und MQTT 5 und zeigt zum Schluss, wie du alles gegen einen echten Broker testest.

Die passende Bibliothek wählen

Für Mikrocontroller ist keine der beiden die übliche Wahl; Embedded-C-Bibliotheken wie coreMQTT oder der ESP-IDF-Client sind schlanker. Der Rest dieses Guides zielt auf Linux-, macOS- und Windows-Hosts. Einen breiteren Überblick über Client-Bibliotheken pro Sprache bietet die besten MQTT-Bibliotheken.

Paho C++ installieren

Paho C++ braucht die Paho-C-Bibliothek (mit SSL gebaut, wenn du TLS willst). Drei gängige Wege:

# Debian / Ubuntu (distribution packages can lag behind the latest release)
sudo apt install libpaho-mqtt-dev libpaho-mqttpp-dev

# vcpkg
vcpkg install paho-mqttpp3

# From source, building the bundled Paho C library with TLS
git clone https://github.com/eclipse/paho.mqtt.cpp
cd paho.mqtt.cpp && git submodule update --init
cmake -Bbuild -DPAHO_WITH_MQTT_C=ON -DPAHO_WITH_SSL=ON
sudo cmake --build build --target install

Eine minimale CMakeLists.txt linkt das importierte Target, das die C-Bibliothek und OpenSSL automatisch mitbringt:

cmake_minimum_required(VERSION 3.16)
project(mqtt_demo CXX)
set(CMAKE_CXX_STANDARD 17)

find_package(PahoMqttCpp REQUIRED)

add_executable(publish publish.cpp)
target_link_libraries(publish PRIVATE PahoMqttCpp::paho-mqttpp3)

add_executable(subscribe subscribe.cpp)
target_link_libraries(subscribe PRIVATE PahoMqttCpp::paho-mqttpp3)

Wenn du von Hand linkst und undefined reference to MQTTAsync_create siehst, fehlt die C-Bibliothek: Linke paho-mqtt3as (die asynchrone Variante mit SSL) nach paho-mqttpp3.

Eine Nachricht publishen

Jeder asynchrone Aufruf liefert ein Token. Mit ->wait() wird daraus ein blockierender Aufruf – genau das, was du in einem kleinen Tool willst:

#include <chrono>
#include <iostream>
#include <mqtt/async_client.h>

int main() {
    const std::string server = "tcp://broker.connectmq.com:1883";
    mqtt::async_client client(server, "testmqtt-cpp-pub");

    auto opts = mqtt::connect_options_builder()
        .clean_session()
        .keep_alive_interval(std::chrono::seconds(30))
        .finalize();

    try {
        client.connect(opts)->wait();

        auto msg = mqtt::make_message("testmqtt/cpp/hello", "Hello from C++", 1, false);
        client.publish(msg)->wait();   // returns after PUBACK for QoS 1
        std::cout << "published" << std::endl;

        client.disconnect()->wait();
    } catch (const mqtt::exception& e) {
        std::cerr << e.what() << std::endl;
        return 1;
    }
}

Bei QoS 1 kehrt wait() auf dem Publish-Token erst zurück, nachdem der Broker die Nachricht bestätigt hat. Bei QoS 0 kehrt es zurück, sobald das Paket geschrieben ist. In MQTT-QoS-Stufen liest du, wann sich welche Stufe lohnt.

Abonnieren mit der Consumer-Queue

Der einfachste Subscriber kommt ganz ohne Callbacks aus: start_consuming() legt eingehende Nachrichten in eine threadsichere Queue, die du in deiner eigenen Schleife ausliest.

#include <iostream>
#include <mqtt/async_client.h>

int main() {
    mqtt::async_client client("tcp://broker.connectmq.com:1883", "testmqtt-cpp-sub");

    auto opts = mqtt::connect_options_builder()
        .clean_session()
        .finalize();

    try {
        client.start_consuming();          // before connect, so nothing is missed
        client.connect(opts)->wait();
        client.subscribe("testmqtt/cpp/#", 1)->wait();
        std::cout << "waiting for messages..." << std::endl;

        while (true) {
            auto msg = client.consume_message();   // blocks
            if (!msg) break;                       // nullptr: connection lost
            std::cout << msg->get_topic() << " -> " << msg->to_string() << std::endl;
        }
    } catch (const mqtt::exception& e) {
        std::cerr << e.what() << std::endl;
        return 1;
    }
}

Starte den Subscriber und dann den Publisher in einem zweiten Terminal. Du kannst auch über den Online-MQTT-Client auf testmqtt/cpp/hello publishen und zusehen, wie die Zeile in deinem Terminal erscheint.

Callbacks und automatischer Reconnect

Langlaufende Dienste bevorzugen meist Callbacks und müssen Netzwerkabbrüche überstehen. Paho C++ hat Lambda-Setter für jedes Ereignis, und automatic_reconnect() übernimmt das Back-off:

#include <chrono>
#include <iostream>
#include <thread>
#include <mqtt/async_client.h>

int main() {
    mqtt::async_client client("tcp://broker.connectmq.com:1883", "testmqtt-cpp-service");

    client.set_connected_handler([&client](const std::string&) {
        std::cout << "connected, subscribing" << std::endl;
        client.subscribe("testmqtt/cpp/commands/#", 1);   // no wait() in a callback!
    });
    client.set_connection_lost_handler([](const std::string& cause) {
        std::cerr << "connection lost: " << cause << std::endl;
    });
    client.set_message_callback([](mqtt::const_message_ptr msg) {
        std::cout << msg->get_topic() << " " << msg->get_payload_str() << std::endl;
    });

    auto opts = mqtt::connect_options_builder()
        .clean_session(false)
        .keep_alive_interval(std::chrono::seconds(30))
        .automatic_reconnect(std::chrono::seconds(1), std::chrono::seconds(30))
        .finalize();

    client.connect(opts)->wait();
    while (true) std::this_thread::sleep_for(std::chrono::seconds(60));
}

Drei Regeln halten Callback-Code gesund:

• Blockiere nie in einem Callback. Ein ->wait() in connected oder message_arrived wartet auf genau dem Thread, der das Token abschließen soll. Stoße den Aufruf an und mach weiter, oder schiebe die Arbeit in deine eigene Queue.
• Beim Verbinden neu abonnieren. Der Connected-Handler läuft nach jedem erfolgreichen (Re-)Connect, sodass die Subscriptions zurückkommen, selbst wenn der Broker die Session verloren hat.
• Verwende eine stabile, eindeutige Client-ID mit clean_session(false). Zwei Prozesse mit derselben ID werfen sich gegenseitig in einer Schleife raus. Erzeuge eine mit dem Client-ID-Generator und lies Best Practices für Reconnects für Hinweise zum Back-off.

Persistente Sessions brauchen einen Broker, der sie speichert: Der gemeinsame Endpoint broker.connectmq.com hält keine Offline-Sessions vor, teste clean_session(false) also gegen deinen eigenen privaten Broker. Auf dem gemeinsamen Endpoint hält das erneute Abonnieren im Connected-Handler den Dienst am Laufen.

TLS und Authentifizierung

Für alles jenseits eines kurzen Tests verbindest du dich per TLS auf Port 8883 mit Zugangsdaten. Stelle das URI-Schema auf ssl:// um und ergänze ssl_options:

#include <cstdlib>
#include <mqtt/async_client.h>

static std::string env(const char* name) {
    const char* v = std::getenv(name);
    return v ? v : "";
}

int main() {
    mqtt::async_client client("ssl://broker.connectmq.com:8883", "testmqtt-cpp-tls");

    auto ssl = mqtt::ssl_options_builder()
        .trust_store("/etc/ssl/certs/ca-certificates.crt")   // system CA bundle
        .enable_server_cert_auth(true)
        .finalize();

    auto opts = mqtt::connect_options_builder()
        .user_name(env("MQTT_USER"))
        .password(env("MQTT_PASS"))
        .ssl(std::move(ssl))
        .finalize();

    client.connect(opts)->wait();
    client.publish("testmqtt/cpp/secure", "over TLS", 1, false)->wait();
    client.disconnect()->wait();
}

Der Pfad zum CA-Bundle unterscheidet sich je nach System (/etc/pki/tls/certs/ca-bundle.crt auf RHEL-Systemen; unter macOS exportierst du eines oder nutzt das Bundle des Homebrew-OpenSSL). Scheitert die Verbindung vor dem CONNACK, prüfe die Zertifikatskette von derselben Maschine aus mit den Schritten aus MQTT-TLS-Verbindung testen.

MQTT 5 mit Paho C++

MQTT 5 musst du explizit aktivieren. Erzeuge den Client mit MQTT-5-Optionen und nutze den v5-Builder, der auch Properties annimmt:

mqtt::async_client client("tcp://broker.connectmq.com:1883", "testmqtt-cpp-v5",
                          mqtt::create_options(MQTTVERSION_5));

auto opts = mqtt::connect_options_builder::v5()
    .clean_start(false)
    .properties({{mqtt::property::SESSION_EXPIRY_INTERVAL, 3600}})
    .finalize();

client.connect(opts)->wait();

Mit MQTT 5 liefert der Broker Reason Codes statt der alten Ein-Byte-Return-Codes, sodass dir ein Fehler deutlich mehr verrät (zum Beispiel 0x87 Not authorized oder 0x97 Quota exceeded). Schlag jeden Code im MQTT-Reason-Codes-Tool nach. Die Unterschiede zwischen den Versionen fasst MQTT 5 vs. 3.1.1 zusammen.

async_mqtt: die Asio-native Option

async_mqtt ist eine Header-only-Bibliothek für C++17 auf Basis von Boost.Asio, vom Autor des älteren mqtt_cpp. Jede Operation nimmt ein Asio Completion Token, sodass derselbe Aufruf einen Callback, ein std::future oder co_await in einer C++20-Coroutine verwenden kann. Sie unterstützt MQTT 3.1.1 und 5 über TCP, TLS und WebSocket und liefert außerdem die Bausteine für einen Broker oder Server-Endpoint.

Wähle sie, wenn:

• dein Programm bereits einen io_context betreibt (ein Asio-basierter Server, ein Gateway mit vielen Sockets),
• du Coroutines statt Callback-Ketten willst,
• du eine Header-only-Abhängigkeit einer kompilierten C-Bibliothek vorziehst.

Die API hat sich zwischen Major-Versionen verändert. Starte daher mit den Beispielen im Repository, die zu deiner installierten Version passen, statt mit älteren Snippets. Konzeptionell ist der Ablauf derselbe wie bei Paho: Transport öffnen, CONNECT senden und auf CONNACK warten, SUBSCRIBE senden und dann in einer Schleife PUBLISH-Pakete empfangen.

Häufige Fehler

End-to-End testen

Bevor du deinem Code die Schuld gibst, beweise mit einem bewährten Client, dass die Broker-Seite funktioniert. Starte einen Subscriber im Terminal und führe deinen Publisher aus:

mosquitto_sub -h broker.connectmq.com -p 1883 -t 'testmqtt/cpp/#' -v

Dann umgekehrt: Starte deinen C++-Subscriber und sende eine Nachricht von der Kommandozeile:

mosquitto_pub -h broker.connectmq.com -p 1883 -t 'testmqtt/cpp/commands/led' -m 'on' -q 1

Um Reconnects zu testen, deaktivierst du dein Netzwerkinterface länger als das Keep-Alive-Intervall und prüfst, ob der Connection-Lost-Handler feuert, der Client sich neu verbindet und wieder Nachrichten fließen. Brauchst du Boilerplate in einer anderen Sprache für die Gegenseite des Tests, erzeugt der MQTT-Codegenerator ihn für dich.

C++-Topics im Online-Client beobachten

Paho C++ liefert dir mit sehr wenig Code einen verlässlichen MQTT-Client: einen Builder für die Optionen, Tokens für den Kontrollfluss und Callbacks oder eine Consumer-Queue für eingehende Nachrichten. Halte Callbacks nicht-blockierend, aktiviere den automatischen Reconnect mit stabilen Client-IDs, setze von Anfang an auf TLS und greif zu async_mqtt, wenn dein Programm ohnehin eine Asio-Anwendung ist.

Häufig gestellte Fragen