MQTT en PHP con php-mqtt/client

La forma estándar de usar MQTT en PHP es la librería php-mqtt/client: la instalas con Composer, creas un MqttClient, llamas a connect() y luego a publish() desde cualquier sitio y a subscribe() con un closure dentro de un bucle CLI de larga duración. Habla MQTT 3.1 y 3.1.1 sobre TCP o TLS, no necesita extensiones de PHP más allá de las predeterminadas y tiene un wrapper oficial para Laravel.

Este tutorial cubre la publicación (incluido el detalle de QoS 1 que casi todo el mundo pasa por alto), un suscriptor robusto, TLS con credenciales, Last Will y cómo probar cada pieza.

Instalación

php-mqtt/client requiere PHP 8 y Composer:

composer require php-mqtt/client

Hay dos formas muy distintas en que PHP habla con un broker, y confundirlas causa la mayoría de los problemas:

Los workers de PHP-FPM se reciclan y las peticiones expiran, así que nunca intentes mantener una suscripción abierta dentro de una petición web.

Publicar un mensaje

<?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";

Dos detalles importan:

• El client ID aleatorio. Cada petición es una conexión distinta. Si dos peticiones simultáneas usaran el mismo ID fijo, el broker desconectaría a la primera; consulta client ID duplicado. El generador de client ID muestra buenos formatos.
• loop(true, true) después de publicar con QoS 1 o 2. publish() solo encola el mensaje; el acuse se gestiona dentro del bucle. El segundo argumento le dice a loop() que vuelva en cuanto no quede nada pendiente. Con QoS 0 puedes desconectar directamente. Los niveles de QoS de MQTT explican el compromiso.

Abrir una conexión por petición está bien para eventos ocasionales. Si una página publica en cada visita, o envías ráfagas de mensajes, pasa el trabajo a un worker de cola que mantenga una conexión abierta: es el patrón que se muestra en MQTT en Laravel.

Mensajes retenidos para el estado

El cuarto argumento de publish() es el flag retain. El broker guarda un mensaje retenido y lo entrega de inmediato a cada nuevo suscriptor, lo que lo hace ideal para topics de "estado actual", como la configuración de un dispositivo que se edita desde un panel de administración en PHP:

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

Usa retain para estados, no para eventos: un mensaje retenido de "botón pulsado" se reproduciría a cada cliente que se suscriba más tarde. Mantén los payloads como JSON compacto con un timestamp, para que los consumidores sepan lo antiguo que es el estado.

Suscribirse en un worker CLI

Un suscriptor registra un closure por cada filtro de topic y luego llama a loop(true), que bloquea y despacha mensajes hasta que lo interrumpas:

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

El comodín + coincide con un nivel de topic, así que este filtro recibe testmqtt/php/sensors/room1 pero no testmqtt/php/sensors/room1/battery. Comprueba tus filtros con el topic matcher antes de desplegar.

Ejecuta php subscriber.php en una terminal y php publish.php en otra. Deberías ver la lectura impresa en milisegundos.

Sobrevivir a las desconexiones

Un worker que funciona durante semanas acabará perdiendo la conexión: reinicios del broker, timeouts de NAT, despliegues. Cuando eso pasa, la librería lanza una excepción desde loop(). El patrón robusto más sencillo es un bucle exterior que se reconecta con back-off y vuelve a suscribirse:

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

El Last Will publica offline (retenido) si el worker muere sin una desconexión limpia, y el worker lo sobrescribe con online tras cada conexión: un indicador de salud barato al que otros servicios pueden suscribirse. Last Will en MQTT cubre los detalles.

Usar aquí una sesión limpia es deliberado: el worker se vuelve a suscribir en cada conexión, así que no depende de que el broker guarde nada. Si necesitas que los mensajes se encolen mientras el worker está offline, cambia a connect($settings, false) con un client ID fijo y pruébalo contra tu propio broker privado: el endpoint compartido broker.connectmq.com no conserva sesiones offline. Consulta sesiones persistentes.

php-mqtt/client también ofrece setReconnectAutomatically(true) con setMaxReconnectAttempts() y setDelayBetweenReconnectAttempts() (en milisegundos), útil para publicadores que mantienen una conexión abierta entre jobs.

TLS y autenticación

Los brokers de producción exigen TLS en el puerto 8883 y credenciales. Solo cambian el puerto y los ajustes:

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

Sin un fichero de CA explícito, el OpenSSL de PHP usa el almacén de confianza del sistema (u openssl.cafile en php.ini). Mantén activada la verificación del peer; desactivarla oculta problemas reales de certificados.

Usarlo en Laravel

Los mismos autores mantienen php-mqtt/laravel-client, que lee los ajustes de conexión de config/mqtt-client.php y .env y expone una facade:

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

Para suscribirte llamas a MQTT::connection() dentro de un comando Artisan y ejecutas loop(true) ahí. El patrón completo (jobs en cola para publicar, un comando supervisado para suscribirse, y tests) está en MQTT en Laravel.

Errores comunes

Pruébalo

Observa lo que publica tu código PHP con un suscriptor en la terminal o con el cliente MQTT online:

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

Después alimenta a tu suscriptor con un mensaje, incluido uno inválido para comprobar que la validación JSON funciona:

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

Por último, mata el worker con kill -9 y confirma que el estado retenido offline aparece en testmqtt/php/worker/status. Tienes más recetas de línea de comandos en ejemplos de mosquitto_pub y mosquitto_sub.

Ver los topics de PHP en el cliente online

php-mqtt/client hace que MQTT en PHP sea sencillo en cuanto respetas el modelo de procesos de PHP: publica desde las peticiones con client IDs únicos y un loop(true, true) para QoS 1, suscríbete solo desde workers CLI supervisados que se reconecten por sí mismos, y activa TLS y credenciales antes de salir a producción.

Preguntas frecuentes