PHP MQTT Client Tutorial mit php-mqtt/client

Mit einem PHP MQTT Client kann Ihre bestehende Webanwendung Befehle an Geräte senden und auf deren Daten reagieren, ohne dass Ihr Stack eine neue Sprache braucht. In diesem Tutorial nutzen Sie das Paket php-mqtt/client, um aus einem Web-Request zu publizieren, betreiben einen dauerhaft laufenden Subscriber-Worker, sichern beides mit TLS ab und halten den Worker in der Produktion am Leben.

Zwei sehr unterschiedliche Aufgaben für einen PHP MQTT Client

Das Request-Modell von PHP bestimmt, wie Sie MQTT einsetzen, trennen Sie die beiden Fälle also von Anfang an:

Der häufigste Fehler ist der Versuch, innerhalb eines Web-Requests zu abonnieren. PHP-FPM-Worker sollen in Millisekunden fertig sein; ein Abonnement braucht einen Prozess, der tagelang verbunden bleibt. Betreiben Sie Subscriber als CLI-Prozess unter der Aufsicht von systemd, Supervisor oder Ihrer Container-Runtime, und beschränken Sie Web-Requests auf kurze Publishes.

php-mqtt/client installieren

php-mqtt/client ist eine reine PHP-Implementierung von MQTT 3.1 und 3.1.1 mit TLS-Unterstützung, wird auf GitHub gepflegt und über Composer installiert. Sie braucht keine PHP-Erweiterung über die standardmäßige Socket- und OpenSSL-Unterstützung hinaus, die die meisten Builds enthalten.

composer require php-mqtt/client

Die README des Pakets ist die offizielle Dokumentation von php-mqtt/client und listet jede Verbindungseinstellung auf. Einen Vergleich mit Clients in anderen Sprachen finden Sie im Leitfaden zu MQTT-Client-Bibliotheken.

Gemeinsame Verbindungseinstellungen

Bündeln Sie die Verbindungsdetails an einer Stelle. Das Beispiel zielt auf den öffentlichen MQTT-Broker unter broker.connectmq.com, der für Tests und Prototypen gedacht ist, über TLS auf Port 8883 und wechselt automatisch zu privaten Zugangsdaten, sobald Umgebungsvariablen gesetzt sind.

mqtt.php:

<?php

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

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

const MQTT_HOST = 'broker.connectmq.com';
const MQTT_PORT = 8883;

function mqtt_topic_prefix(): string
{
    return getenv('MQTT_TOPIC_PREFIX') ?: 'connectmq/demo/php-5e19c2';
}

function mqtt_client(string $role): MqttClient
{
    $clientId = getenv('MQTT_CLIENT_ID') ?: sprintf('php-%s-%s', $role, bin2hex(random_bytes(4)));
    return new MqttClient(MQTT_HOST, MQTT_PORT, $clientId, MqttClient::MQTT_3_1_1);
}

function mqtt_settings(bool $reconnect = false): ConnectionSettings
{
    $settings = (new ConnectionSettings())
        ->setUseTls(true)
        ->setTlsVerifyPeer(true)
        ->setTlsVerifyPeerName(true)
        ->setConnectTimeout(10)
        ->setSocketTimeout(5)
        ->setKeepAliveInterval(30);

    if ($user = getenv('MQTT_USERNAME')) {
        $settings = $settings
            ->setUsername($user)
            ->setPassword(getenv('MQTT_PASSWORD') ?: null);
    }

    if ($reconnect) {
        $settings = $settings
            ->setReconnectAutomatically(true)
            ->setMaxReconnectAttempts(10)
            ->setDelayBetweenReconnectAttempts(3000);
    }

    return $settings;
}

ConnectionSettings ist unveränderlich: Jeder Setter gibt eine neue Instanz zurück, deshalb weist der Code $settings neu zu. Die TLS-Prüfung nutzt den CA-Store des Systems, der dem Let's-Encrypt-Zertifikat von ConnectMQ bereits vertraut.

Aus einem Web-Request publizieren

Dieser Endpunkt nimmt ein Formular entgegen und publiziert einen Befehl an ein Gerät. Mit QoS 1 muss der Broker die Nachricht bestätigen, und php-mqtt/client verarbeitet diese Bestätigung nur innerhalb seiner Schleife. Deshalb lässt der Code die Schleife laufen, bis die ausgehende Queue leer ist, bevor er trennt.

public/command.php:

<?php

require __DIR__ . '/../mqtt.php';

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

$device = preg_replace('/[^a-z0-9_-]/i', '', $_POST['device'] ?? '');
$action = in_array($_POST['action'] ?? '', ['on', 'off'], true) ? $_POST['action'] : null;

if ($device === '' || $action === null) {
    http_response_code(400);
    exit('Invalid device or action');
}

$topic = mqtt_topic_prefix() . "/devices/{$device}/set";
$payload = json_encode(['action' => $action, 'ts' => time()]);

try {
    $mqtt = mqtt_client('web');
    $mqtt->connect(mqtt_settings(), true);
    $mqtt->publish($topic, $payload, MqttClient::QOS_AT_LEAST_ONCE);
    $mqtt->loop(true, true);   // warten, bis das PUBACK empfangen wurde
    $mqtt->disconnect();
    echo "Sent {$action} to {$device}";
} catch (MqttClientException $e) {
    error_log('MQTT publish failed: ' . $e->getMessage());
    http_response_code(502);
    echo 'Device gateway unavailable';
}

Beachten Sie die Bereinigung der Eingabe bei device. Nutzer dürfen niemals #, + oder / in einen Topic-Namen einschleusen können, sonst können sie Geräte ansprechen, die ihnen nicht gehören.

Eine TLS-Verbindung pro Request kostet Latenz, typischerweise ein bis zwei Handshake-Round-Trips. Für ein Admin-Panel, das einige Befehle pro Minute sendet, ist das unproblematisch. Für Publishing mit hohem Volumen aus PHP schieben Sie die Arbeit in eine Queue (Redis, eine Datenbanktabelle, Laravel Queues) und lassen einen dauerhaft laufenden Worker mit einer einzigen MQTT-Verbindung sie abarbeiten.

Ein dauerhaft laufender PHP MQTT Subscriber

Der Subscriber ist ein CLI-Skript. Er abonniert mit einer Wildcard, dekodiert JSON und beendet sich bei SIGTERM sauber, sodass systemd oder Docker ihn neu starten können, ohne eine Nachricht mitten in der Verarbeitung zu verlieren.

bin/subscriber.php:

<?php

require __DIR__ . '/../mqtt.php';

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

pcntl_async_signals(true);

$mqtt = mqtt_client('worker');

$stop = function () use ($mqtt) {
    fwrite(STDOUT, "Shutting down...\n");
    $mqtt->interrupt();
};
pcntl_signal(SIGINT, $stop);
pcntl_signal(SIGTERM, $stop);

$filter = mqtt_topic_prefix() . '/devices/+/telemetry';

try {
    $mqtt->connect(mqtt_settings(reconnect: true), true);

    $mqtt->subscribe($filter, function (string $topic, string $message, bool $retained) {
        $parts = explode('/', $topic);
        $device = $parts[count($parts) - 2] ?? 'unknown';
        $data = json_decode($message, true);

        if (!is_array($data)) {
            fwrite(STDERR, "Ignoring non-JSON payload on {$topic}\n");
            return;
        }

        // Durch eigene Persistenz ersetzen, z. B. PDO-Insert oder ein Framework-Model
        printf("[%s] %s%s %s\n", date('c'), $device, $retained ? ' (retained)' : '', json_encode($data));
    }, MqttClient::QOS_AT_LEAST_ONCE);

    fwrite(STDOUT, "Subscribed to {$filter}\n");
    $mqtt->loop(true);
    $mqtt->disconnect();
} catch (MqttClientException $e) {
    fwrite(STDERR, 'MQTT error: ' . $e->getMessage() . "\n");
    exit(1);
}

Starten Sie ihn:

php bin/subscriber.php

Spielen Sie in einem anderen Terminal ein Gerät:

mosquitto_pub -h broker.connectmq.com -p 1883 -i php-sim-pump1 \
  -t "connectmq/demo/php-5e19c2/devices/pump1/telemetry" \
  -m '{"flow": 12.4, "pressure": 2.1}'

Die Erweiterung pcntl ist im PHP-CLI unter Linux verfügbar, in der Regel aber nicht unter PHP-FPM oder unter Windows, ein weiterer Grund, den Subscriber als CLI-Prozess zu betreiben.

Den Worker in der Produktion betreiben

/etc/systemd/system/php-mqtt-worker.service:

[Unit]
Description=PHP MQTT subscriber
After=network-online.target
Wants=network-online.target

[Service]
User=www-data
WorkingDirectory=/var/www/app
ExecStart=/usr/bin/php bin/subscriber.php
EnvironmentFile=/etc/php-mqtt-worker.env
Restart=always
RestartSec=5
KillSignal=SIGTERM

[Install]
WantedBy=multi-user.target

/etc/php-mqtt-worker.env (Modus 600) enthält die Zugangsdaten des privaten Brokers. Gehört die Webanwendung zu einem MES- oder Instandhaltungsportal, richten Sie MQTT_TOPIC_PREFIX an der ISA-95-Struktur Ihres Unified Namespace (UNS) aus, etwa acme/werk-stuttgart:

MQTT_USERNAME=YOUR_USERNAME
MQTT_PASSWORD=YOUR_PASSWORD
MQTT_CLIENT_ID=YOUR_CLIENT_ID
MQTT_TOPIC_PREFIX=acme/site1

Lang laufende PHP-Prozesse haben ihre eigenen Fehlerbilder. Gehen Sie diese Checkliste durch, bevor Sie sich auf den Worker verlassen:

1. Speicher. Geben Sie große Arrays mit unset frei, sammeln Sie keine Daten in statischen Properties an und erwägen Sie, den Prozess nach einigen Stunden zu beenden, damit systemd einen sauberen Prozess startet.
2. Datenbankverbindungen. PDO-Verbindungen laufen ab, während der Worker untätig ist; verbinden Sie sich bei Fehlern neu, statt eine Verbindung ewig zu halten.
3. Eindeutige Client IDs. Der Web-Publisher und der Worker brauchen unterschiedliche Client IDs. Teilen sich zwei Prozesse eine, trennt der Broker den älteren, und beide werfen sich gegenseitig in einer Schleife hinaus.
4. Reconnect-Limits. Mit setMaxReconnectAttempts(10) wirft die Schleife irgendwann eine Exception; lassen Sie den Prozess mit einem Exit-Code ungleich null enden, damit der Supervisor ihn neu startet.
5. Persistente Sessions. Darf der Worker bei Neustarts keine Nachrichten verpassen, verbinden Sie sich mit deaktivierter Clean Session und einer festen Client ID und abonnieren Sie mit QoS 1. Der Leitfaden zu persistenten Sessions erklärt die Abwägung.

Auf einem privaten ConnectMQ Broker sind Client IDs an die ausgegebenen Zugangsdaten gebunden. Planen Sie also einen Satz Zugangsdaten pro gleichzeitig verbundenem Prozess ein.

QoS, Retain und Last Will in PHP wählen

Für Befehle aus einer Web-App ist QoS 1 der sinnvolle Standard: Der Broker bestätigt den Empfang, und das Gerät sollte den resultierenden Zustand auf seinem eigenen Topic melden, damit die UI zeigt, was wirklich passiert ist. Vermeiden Sie das Retain-Flag bei Befehlen. Ein retained on würde dem Gerät bei jedem Reconnect erneut zugestellt, womöglich Stunden später; Retain ist für Zustände gedacht, nicht für Aktionen, wie der Leitfaden zu MQTT Retained Messages erklärt.

Der Subscriber-Worker kann seinen eigenen Zustand per Last Will melden. Ergänzen Sie die Verbindungseinstellungen des Workers um ->setLastWillTopic(mqtt_topic_prefix() . '/workers/php/status'), ->setLastWillMessage('offline'), ->setLastWillQualityOfService(1) und ->setRetainLastWill(true) und publizieren Sie direkt nach dem Verbinden ein retained online. Stürzt der Worker ab oder verliert er sein Netzwerk, publiziert der Broker für ihn offline, und Ihr Monitoring kann auf dieses Topic hin alarmieren, statt aus Stille Schlüsse zu ziehen. Der kostenlose ConnectMQ Serverless-Plan unterstützt QoS 0 und 1 sowie Nachrichten bis 256 KB, was Befehls- und Telemetrie-Payloads aus einer PHP-Anwendung bequem abdeckt.

Hinweise zu Laravel und Symfony

Derselbe Code lässt sich in Frameworks übernehmen. In Laravel kapseln Sie den Subscriber in einen Artisan-Befehl (php artisan mqtt:listen), betreiben ihn unter Supervisor und publizieren aus einem Queued Job statt aus dem Controller. In Symfony übernimmt ein Console Command dieselbe Rolle. Das Community-Paket php-mqtt/laravel-client ergänzt Konfiguration und eine Facade auf Basis derselben Bibliothek, falls Sie diesen Stil bevorzugen.

Fehlerbehebung

Starten Sie mit ConnectMQ

Geben Sie Ihrer PHP-Anwendung in wenigen Minuten einen privaten, sicheren Broker. Erstellen Sie unter app.connectmq.com ohne Kreditkarte einen kostenlosen Serverless Broker, tragen Sie die generierten Zugangsdaten in Ihre Umgebungsdatei ein und testen Sie Ihre Publishes mit dem Browser-Client des Portals. Mit einem verwalteten MQTT-Broker bleiben TLS, Skalierung und Broker-Upgrades vom Tisch Ihres PHP-Teams. Für andere Backend-Sprachen lesen Sie das Ruby MQTT Client Tutorial und das Paho MQTT C Tutorial.

Häufig gestellte Fragen

Unterstützt php-mqtt/client MQTT 5?

Nein. Das Paket unterstützt MQTT 3.1 und 3.1.1, die ConnectMQ neben MQTT 5.0 akzeptiert. Für die meisten Befehlsflüsse vom Web zum Gerät reicht 3.1.1 aus.

Kann ich MQTT-Topics innerhalb eines PHP-Web-Requests abonnieren?

Das ist möglich, aber nicht empfehlenswert. Abonnements brauchen einen Prozess, der verbunden bleibt. Betreiben Sie sie daher in einem CLI-Worker und beschränken Sie Web-Requests auf kurze Publishes.

Warum kommt mein QoS-1-Publish aus PHP manchmal nicht an?

Wahrscheinlich wurde das Skript beendet, bevor der Client die Bestätigung des Brokers verarbeitet hat. Rufen Sie nach dem Publish $mqtt->loop(true, true) auf, damit sich die Queue vor disconnect() leert.

Wie verbinde ich PHP über TLS mit MQTT?

Aktivieren Sie setUseTls(true) in den Verbindungseinstellungen und verbinden Sie sich mit Port 8883. Die Peer-Prüfung funktioniert sofort mit Brokern, die öffentlich vertrauenswürdige Zertifikate nutzen, etwa broker.connectmq.com.