MQTT mit PHP und php-mqtt/client

Der Standardweg zu MQTT mit PHP ist die Bibliothek php-mqtt/client: Mit Composer installieren, einen MqttClient erzeugen, connect() aufrufen, dann von überall publish() und in einer langlaufenden CLI-Schleife subscribe() mit einer Closure. Sie spricht MQTT 3.1 und 3.1.1 über TCP oder TLS, braucht keine PHP-Erweiterung über die Standardausstattung hinaus und hat einen offiziellen Laravel-Wrapper.

Dieses Tutorial behandelt das Publishen (inklusive des QoS-1-Details, das die meisten übersehen), einen robusten Subscriber, TLS mit Zugangsdaten, Last Will und wie du jedes Teil testest.

Installation

php-mqtt/client benötigt PHP 8 und Composer:

composer require php-mqtt/client

PHP spricht auf zwei sehr unterschiedliche Arten mit einem Broker – wer sie verwechselt, verursacht die meisten Probleme:

PHP-FPM-Worker werden recycelt und Requests laufen in Timeouts – versuche also nie, eine Subscription innerhalb eines Web-Requests offen zu halten.

Eine Nachricht publishen

<?php
require __DIR__ . '/vendor/autoload.php';

use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\MqttClient;

$clientId = 'testmqtt-php-' . bin2hex(random_bytes(4));
$mqtt = new MqttClient('broker.connectmq.com', 1883, $clientId, MqttClient::MQTT_3_1_1);

$settings = (new ConnectionSettings)
    ->setConnectTimeout(5)
    ->setKeepAliveInterval(30);

$mqtt->connect($settings, true);          // true = clean session

$payload = json_encode(['temp' => 21.5, 'ts' => time()]);
$mqtt->publish('testmqtt/php/sensors/room1', $payload, MqttClient::QOS_AT_LEAST_ONCE);

$mqtt->loop(true, true);                  // wait for the PUBACK, then return
$mqtt->disconnect();
echo "published\n";

Zwei Details sind entscheidend:

• Die zufällige Client-ID. Jeder Request ist eine eigene Verbindung. Würden zwei gleichzeitige Requests dieselbe feste ID nutzen, würde der Broker den ersten trennen – siehe doppelte Client-ID. Der Client-ID-Generator zeigt gute Formate.
• loop(true, true) nach einem Publish mit QoS 1 oder 2. publish() stellt die Nachricht nur in die Warteschlange; die Bestätigung wird in der Schleife verarbeitet. Das zweite Argument sorgt dafür, dass loop() zurückkehrt, sobald nichts mehr aussteht. Bei QoS 0 kannst du sofort trennen. MQTT-QoS-Stufen erklärt den Kompromiss.

Eine Verbindung pro Request ist für gelegentliche Ereignisse in Ordnung. Publisht eine Seite bei jedem Aufruf oder sendest du Nachrichten in Schüben, verlagere die Arbeit in einen Queue-Worker, der eine Verbindung offen hält – das Muster aus MQTT mit Laravel.

Retained Messages für Zustände

Das vierte Argument von publish() ist das Retain-Flag. Eine Retained Message speichert der Broker und liefert sie jedem neuen Subscriber sofort aus. Das ist ideal für „aktueller Zustand“-Topics wie eine Gerätekonfiguration, die in einem PHP-Adminpanel bearbeitet wird:

$config = ['interval' => 60, 'unit' => 'C'];
$mqtt->publish('testmqtt/php/devices/42/config', json_encode($config), MqttClient::QOS_AT_LEAST_ONCE, true);

// An empty retained payload deletes the stored message
$mqtt->publish('testmqtt/php/devices/42/config', '', MqttClient::QOS_AT_LEAST_ONCE, true);
$mqtt->loop(true, true);

Nutze Retain für Zustände, nicht für Ereignisse: Eine Retained Message „Button gedrückt“ würde jedem Client erneut zugestellt, der später abonniert. Halte Payloads als kompaktes JSON mit Zeitstempel, damit Consumer erkennen, wie alt der Zustand ist.

Abonnieren in einem CLI-Worker

Ein Subscriber registriert pro Topic-Filter eine Closure und ruft dann loop(true) auf. Die Schleife blockiert und verteilt Nachrichten, bis du sie unterbrichst:

<?php
require __DIR__ . '/vendor/autoload.php';

use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\MqttClient;

$mqtt = new MqttClient('broker.connectmq.com', 1883, 'testmqtt-php-worker', MqttClient::MQTT_3_1_1);
$mqtt->connect((new ConnectionSettings)->setKeepAliveInterval(30), true);

$mqtt->subscribe('testmqtt/php/sensors/+', function (string $topic, string $message, bool $retained) {
    $data = json_decode($message, true);
    if (!is_array($data)) {
        fwrite(STDERR, "invalid JSON on $topic\n");
        return;
    }
    printf("%s%s temp=%s\n", $topic, $retained ? ' (retained)' : '', $data['temp'] ?? '?');
}, MqttClient::QOS_AT_LEAST_ONCE);

// Stop cleanly on Ctrl+C or a SIGTERM from systemd / Docker
pcntl_async_signals(true);
pcntl_signal(SIGINT, fn () => $mqtt->interrupt());
pcntl_signal(SIGTERM, fn () => $mqtt->interrupt());

$mqtt->loop(true);
$mqtt->disconnect();

Die Wildcard + passt auf genau eine Topic-Ebene. Dieser Filter empfängt also testmqtt/php/sensors/room1, aber nicht testmqtt/php/sensors/room1/battery. Prüfe Filter vor dem Deployment mit dem Topic Matcher.

Starte php subscriber.php in einem Terminal und php publish.php in einem anderen. Der Messwert sollte innerhalb von Millisekunden erscheinen.

Verbindungsabbrüche überstehen

Ein Worker, der wochenlang läuft, verliert irgendwann seine Verbindung: Broker-Neustarts, NAT-Timeouts, Deployments. Die Bibliothek wirft dann eine Exception aus loop(). Das einfachste robuste Muster ist eine äußere Schleife, die sich mit Back-off neu verbindet und erneut abonniert:

<?php
require __DIR__ . '/vendor/autoload.php';

use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\Exceptions\MqttClientException;
use PhpMqtt\Client\MqttClient;

$settings = (new ConnectionSettings)
    ->setKeepAliveInterval(30)
    ->setLastWillTopic('testmqtt/php/worker/status')
    ->setLastWillMessage('offline')
    ->setLastWillQualityOfService(MqttClient::QOS_AT_LEAST_ONCE)
    ->setRetainLastWill(true);

$delay = 1;
while (true) {
    try {
        $mqtt = new MqttClient('broker.connectmq.com', 1883, 'testmqtt-php-worker', MqttClient::MQTT_3_1_1);
        $mqtt->connect($settings, true);
        $mqtt->publish('testmqtt/php/worker/status', 'online', MqttClient::QOS_AT_LEAST_ONCE, true);
        $mqtt->subscribe('testmqtt/php/sensors/+', function (string $topic, string $message) {
            echo "$topic $message\n";
        }, MqttClient::QOS_AT_LEAST_ONCE);

        $delay = 1;                       // connected: reset back-off
        $mqtt->loop(true);
    } catch (MqttClientException $e) {
        fwrite(STDERR, 'MQTT error: ' . $e->getMessage() . "\n");
        sleep($delay);
        $delay = min($delay * 2, 30);
    }
}

Der Last Will publisht offline (retained), wenn der Worker ohne sauberes Disconnect stirbt, und der Worker überschreibt ihn nach jedem Connect mit online – ein günstiger Health-Indikator, den andere Dienste abonnieren können. Details liefert MQTT Last Will.

Die Clean Session ist hier Absicht: Der Worker abonniert bei jedem Connect neu und hängt nicht davon ab, dass der Broker etwas speichert. Sollen Nachrichten gepuffert werden, während der Worker offline ist, wechsle zu connect($settings, false) mit fester Client-ID und teste das gegen deinen eigenen privaten Broker – der gemeinsame Endpoint broker.connectmq.com hält keine Offline-Sessions vor. Siehe persistente Sessions.

php-mqtt/client bietet außerdem setReconnectAutomatically(true) mit setMaxReconnectAttempts() und setDelayBetweenReconnectAttempts() (Millisekunden) – praktisch für Publisher, die zwischen Jobs eine Verbindung offen halten.

TLS und Authentifizierung

Produktions-Broker verlangen TLS auf Port 8883 und Zugangsdaten. Es ändern sich nur Port und Einstellungen:

<?php
require __DIR__ . '/vendor/autoload.php';

use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\MqttClient;

$settings = (new ConnectionSettings)
    ->setUsername(getenv('MQTT_USER'))
    ->setPassword(getenv('MQTT_PASS'))
    ->setUseTls(true)
    ->setTlsVerifyPeer(true)
    ->setTlsVerifyPeerName(true)
    ->setTlsSelfSignedAllowed(false);
    // ->setTlsCertificateAuthorityFile('/path/to/ca.pem') for a private CA

$mqtt = new MqttClient('broker.connectmq.com', 8883, 'testmqtt-php-' . bin2hex(random_bytes(4)), MqttClient::MQTT_3_1_1);
$mqtt->connect($settings, true);
$mqtt->publish('testmqtt/php/secure', 'hello over TLS', MqttClient::QOS_AT_LEAST_ONCE);
$mqtt->loop(true, true);
$mqtt->disconnect();

Ohne explizite CA-Datei nutzt PHPs OpenSSL den Trust Store des Systems (oder openssl.cafile in der php.ini). Lass die Peer-Verifizierung eingeschaltet; wer sie abschaltet, verdeckt echte Zertifikatsprobleme.

Einsatz in Laravel

Dieselben Autoren pflegen php-mqtt/laravel-client. Das Paket liest die Verbindungseinstellungen aus config/mqtt-client.php und .env und stellt eine Facade bereit:

composer require php-mqtt/laravel-client
php artisan vendor:publish --provider="PhpMqtt\Client\MqttClientServiceProvider" --tag="config"

use PhpMqtt\Client\Facades\MQTT;

MQTT::publish('testmqtt/php/laravel/hello', 'Hello from Laravel');

Zum Abonnieren rufst du MQTT::connection() in einem Artisan-Command auf und startest dort loop(true). Das komplette Muster – Queue-Jobs zum Publishen, ein überwachter Command zum Abonnieren und Tests – findest du in MQTT mit Laravel.

Häufige Fehler

Testen

Beobachte mit einem Terminal-Subscriber oder dem Online-MQTT-Client, was dein PHP-Code publisht:

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

Füttere dann deinen Subscriber mit einer Nachricht – auch mit einer ungültigen, um zu beweisen, dass die JSON-Prüfung greift:

mosquitto_pub -h broker.connectmq.com -p 1883 -t 'testmqtt/php/sensors/room1' -m '{"temp":22.1}' -q 1
mosquitto_pub -h broker.connectmq.com -p 1883 -t 'testmqtt/php/sensors/room1' -m 'not json' -q 1

Beende den Worker zum Schluss mit kill -9 und prüfe, ob der retained Status offline auf testmqtt/php/worker/status erscheint. Weitere Kommandozeilen-Rezepte findest du in mosquitto_pub- und mosquitto_sub-Beispiele.

PHP-Topics im Online-Client beobachten

php-mqtt/client macht MQTT mit PHP unkompliziert, sobald du das Prozessmodell von PHP respektierst: Aus Requests mit eindeutigen Client-IDs und loop(true, true) für QoS 1 publishen, nur aus überwachten CLI-Workern abonnieren, die sich selbst neu verbinden, und TLS sowie Zugangsdaten aktivieren, bevor du live gehst.

Häufig gestellte Fragen