C++ ile MQTT: Paho C++ ve async_mqtt Örnekleri

C++ ile MQTT kullanmanın en hızlı yolu Eclipse Paho C++ istemcisidir (paho.mqtt.cpp): bir mqtt::async_client oluşturur, connect_options_builder ile bağlanır, ardından publish() ve subscribe() çağırırsınız; her çağrı üzerinde wait() yapabileceğiniz bir token döndürür. Uygulamanız zaten Boost.Asio üzerinde çalışıyorsa async_mqtt, bu modele doğal biçimde oturan header-only bir alternatiftir.

Bu eğitimde Paho C++ ile bir publisher ve bir subscriber yazıyor, TLS, yeniden bağlanma ve MQTT 5 ekliyor ve son olarak her şeyi gerçek bir broker'a karşı nasıl test edeceğinizi gösteriyoruz.

Kütüphane seçimi

Mikrodenetleyiciler için ikisi de olağan tercih değildir; coreMQTT veya ESP-IDF istemcisi gibi gömülü C kütüphaneleri daha hafiftir. Bu rehberin geri kalanı Linux, macOS ve Windows host'larını hedefler. Dillere göre istemci kütüphanelerine daha geniş bir bakış için en iyi MQTT kütüphaneleri yazısına göz atın.

Paho C++ kurulumu

Paho C++, Paho C kütüphanesine ihtiyaç duyar (TLS istiyorsanız SSL ile derlenmiş olmalı). Üç yaygın yol:

# 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

Minimal bir CMakeLists.txt, C kütüphanesini ve OpenSSL'i sizin yerinize getiren import edilmiş target'ı bağlar:

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)

Elle link'liyorsanız ve undefined reference to MQTTAsync_create görüyorsanız C kütüphanesi eksiktir: paho-mqttpp3'ten sonra paho-mqtt3as (SSL destekli asenkron varyant) ekleyin.

Mesaj publish etme

Her asenkron çağrı bir token döndürür. ->wait() çağırmak onu bloklayan bir çağrıya dönüştürür; küçük bir araçta tam da istediğiniz budur:

#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;
    }
}

QoS 1'de publish token'ı üzerindeki wait(), ancak broker mesajı onayladıktan sonra döner. QoS 0'da paket yazılır yazılmaz döner. Hangisinin ne zaman değdiğini görmek için MQTT QoS seviyeleri yazısına bakın.

Consumer kuyruğu ile subscribe

En basit subscriber callback'leri tamamen atlar: start_consuming(), gelen mesajları kendi döngünüzde okuduğunuz thread-safe bir kuyruğa koyar.

#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;
    }
}

Önce subscriber'ı, ardından ikinci bir terminalde publisher'ı çalıştırın. Online MQTT istemcisi üzerinden testmqtt/cpp/hello topic'ine publish edip satırın terminalinizde belirdiğini de görebilirsiniz.

Callback'ler ve otomatik yeniden bağlanma

Uzun süre çalışan servisler genellikle callback'leri tercih eder ve ağ kesintilerini atlatmak zorundadır. Paho C++ her olay için lambda setter'ları sunar; automatic_reconnect() ise back-off'u yönetir:

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

Callback kodunu sağlıklı tutan üç kural:

• Callback içinde asla bloklamayın. connected veya message_arrived içinde ->wait() çağırmak, token'ı tamamlaması gereken thread'in ta kendisinde bekler. Çağrıyı başlatıp devam edin ya da işi kendi kuyruğunuza aktarın.
• Bağlanınca yeniden abone olun. Connected handler her başarılı (yeniden) bağlantıdan sonra çalışır; böylece broker oturumu kaybetmiş olsa bile abonelikler geri gelir.
• clean_session(false) ile sabit ve benzersiz bir client ID kullanın. Aynı ID'ye sahip iki süreç birbirini döngü halinde düşürür. Client ID oluşturucu ile bir tane üretin ve back-off için yeniden bağlanma en iyi uygulamaları yazısını okuyun.

Kalıcı oturumlar, onları saklayan bir broker gerektirir: paylaşımlı broker.connectmq.com endpoint'i çevrimdışı oturum tutmaz; bu yüzden clean_session(false) davranışını kendi özel broker'ınızda test edin. Paylaşımlı endpoint'te servisi ayakta tutan şey connected handler içindeki yeniden aboneliktir.

TLS ve kimlik doğrulama

Hızlı bir testin ötesindeki her şey için kimlik bilgileriyle 8883 portundan TLS üzerinden bağlanın. URI şemasını ssl:// yapın ve ssl_options ekleyin:

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

CA paketinin yolu işletim sistemine göre değişir (RHEL ailesinde /etc/pki/tls/certs/ca-bundle.crt; macOS'ta bir tane dışa aktarın veya Homebrew OpenSSL paketini gösterin). Bağlantı CONNACK'ten önce başarısız oluyorsa, MQTT TLS bağlantısını test etme adımlarıyla aynı makineden sertifika zincirini doğrulayın.

Paho C++ ile MQTT 5

MQTT 5'i açıkça etkinleştirmeniz gerekir. İstemciyi MQTT 5 seçenekleriyle oluşturun ve property de kabul eden v5 builder'ı kullanın:

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

MQTT 5'te broker eski tek baytlık return code'lar yerine reason code döndürür; bu yüzden bir hata size çok daha fazlasını anlatır (örneğin 0x87 Not authorized veya 0x97 Quota exceeded). Herhangi bir kodu MQTT reason code aracı ile sorgulayın. Sürümler arasındaki farklar MQTT 5 ve 3.1.1 yazısında özetleniyor.

async_mqtt: Asio'ya özgü seçenek

async_mqtt, eski mqtt_cpp'nin yazarından, Boost.Asio üzerine kurulu header-only bir C++17 kütüphanesidir. Her işlem bir Asio completion token alır; böylece aynı çağrı bir callback, bir std::future veya bir C++20 coroutine içinde co_await kullanabilir. TCP, TLS ve WebSocket üzerinden MQTT 3.1.1 ve 5'i destekler, ayrıca bir broker veya sunucu endpoint'i için yapı taşları da sağlar.

Şu durumlarda tercih edin:

• programınız zaten bir io_context çalıştırıyorsa (Asio tabanlı bir sunucu, çok sayıda soketi olan bir gateway),
• callback zincirleri yerine coroutine istiyorsanız,
• derlenmiş bir C kütüphanesi yerine header-only bir bağımlılığı tercih ediyorsanız.

API'si major sürümler arasında değişti; bu yüzden eski snippet'ler yerine kurduğunuz sürüme karşılık gelen depo örneklerinden başlayın. Kavramsal olarak akış Paho ile aynıdır: transport'u aç, CONNECT gönder ve CONNACK'i bekle, SUBSCRIBE gönder, ardından döngü içinde PUBLISH paketlerini al.

Sık karşılaşılan hatalar

Uçtan uca test

Kodunuzu suçlamadan önce, güvenilir bir istemciyle broker tarafının çalıştığını kanıtlayın. Terminalde bir subscriber başlatın ve publisher'ınızı çalıştırın:

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

Sonra tersini yapın: C++ subscriber'ınızı çalıştırın ve komut satırından bir mesaj gönderin:

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

Yeniden bağlanmayı denemek için ağ arayüzünüzü keep-alive süresinden daha uzun süre kapatın ve connection-lost handler'ın tetiklendiğini, istemcinin yeniden bağlandığını ve mesajların tekrar aktığını doğrulayın. Testin diğer tarafı için başka bir dilde başlangıç koduna ihtiyacınız varsa MQTT kod oluşturucu sizin için üretir.

C++ topic'lerini online istemcide izleyin

Paho C++ çok az kodla güvenilir bir MQTT istemcisi sunar: seçenekler için bir builder, akış kontrolü için token'lar ve gelen mesajlar için callback'ler ya da bir consumer kuyruğu. Callback'leri bloklamadan tutun, sabit client ID'lerle otomatik yeniden bağlanmayı açın, baştan TLS kullanın ve programınız zaten bir Asio uygulamasıysa async_mqtt'ye yönelin.

Sıkça sorulan sorular