Tutorial de cliente MQTT en PHP con php-mqtt/client

Un cliente MQTT en PHP permite que tu aplicación web actual envíe comandos a los dispositivos y reaccione a sus datos sin añadir un nuevo lenguaje a tu stack. En este tutorial usarás el paquete php-mqtt/client para publicar desde una petición web, ejecutar un worker suscriptor permanente, proteger ambos con TLS y mantener el worker vivo en producción.

Dos trabajos muy distintos para un cliente MQTT en PHP

El modelo de peticiones de PHP condiciona cómo usas MQTT, así que separa los dos casos desde el principio:

Intentar suscribirse dentro de una petición web es el error más común. Los workers de PHP-FPM están pensados para terminar en milisegundos; una suscripción necesita un proceso que permanezca conectado durante días. Mantén los suscriptores en un proceso CLI supervisado por systemd, Supervisor o tu runtime de contenedores, y limita las peticiones web a publicaciones breves.

Instalar php-mqtt/client

php-mqtt/client es una implementación en PHP puro de MQTT 3.1 y 3.1.1 con soporte TLS, mantenida en GitHub e instalada mediante Composer. No necesita ninguna extensión de PHP más allá del soporte estándar de sockets y OpenSSL que incluyen la mayoría de las builds.

composer require php-mqtt/client

El README del paquete es la documentación oficial de php-mqtt/client y enumera todos los ajustes de conexión. Para compararlo con clientes en otros lenguajes, consulta la guía de bibliotecas cliente MQTT.

Ajustes de conexión compartidos

Pon los datos de conexión en un solo lugar. El ejemplo apunta al servidor MQTT público gratuito en broker.connectmq.com (solo para pruebas y prototipos, no para datos de producción) por TLS en el puerto 8883, y cambia automáticamente a credenciales privadas cuando se definen las variables de entorno.

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 es inmutable: cada setter devuelve una instancia nueva, por eso el código vuelve a asignar $settings. La verificación TLS usa el almacén de CA del sistema, que ya confía en el certificado de Let's Encrypt que sirve ConnectMQ.

Publicar desde una petición web

Este endpoint recibe el envío de un formulario y publica un comando hacia un dispositivo. Con QoS 1 el broker debe confirmar el mensaje, y php-mqtt/client solo procesa esa confirmación dentro de su bucle, así que el código ejecuta el bucle hasta que la cola de salida queda vacía antes de desconectar.

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);   // esperar hasta recibir el PUBACK
    $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';
}

Fíjate en la sanitización de la entrada device. Los usuarios nunca deben poder inyectar #, + o / en el nombre de un topic, o podrían dirigirse a dispositivos que no son suyos.

Una conexión TLS por petición añade latencia, normalmente uno o dos viajes de ida y vuelta del handshake. Para un panel de administración que envía unos pocos comandos por minuto, no supone un problema. Para publicar grandes volúmenes desde PHP, pasa el trabajo a una cola (Redis, una tabla de base de datos, las colas de Laravel) y deja que un worker permanente mantenga una única conexión MQTT y la vacíe.

Un suscriptor MQTT en PHP permanente

El suscriptor es un script CLI. Se suscribe con un comodín, decodifica JSON y se detiene limpiamente con SIGTERM para que systemd o Docker puedan reiniciarlo sin perder un mensaje a mitad de procesamiento.

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

        // Sustitúyelo por tu propia persistencia, p. ej. un insert con PDO o un modelo del framework
        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);
}

Ejecútalo:

php bin/subscriber.php

En otra terminal, haz de dispositivo:

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}'

La extensión pcntl está disponible en el CLI de PHP en Linux, pero normalmente no bajo PHP-FPM ni en Windows, lo que es una razón más para mantener el suscriptor como proceso CLI.

Ejecutar el worker en producción

/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 (modo 600) contiene las credenciales del broker privado:

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

Los procesos PHP de larga duración tienen sus propios modos de fallo. Repasa esta lista antes de confiar en el worker:

1. Memoria. Libera los arrays grandes, evita acumular datos en propiedades estáticas y plantéate salir tras unas horas para que systemd reinicie un proceso limpio.
2. Conexiones a la base de datos. Las conexiones PDO caducan mientras el worker está inactivo; reconecta ante un fallo en lugar de mantener una conexión para siempre.
3. Client ID únicos. El publicador web y el worker necesitan client ID distintos. Si dos procesos comparten uno, el broker desconecta el más antiguo y se expulsan mutuamente en bucle.
4. Límites de reconexión. Con setMaxReconnectAttempts(10) el bucle acaba lanzando una excepción; deja que salga con un código distinto de cero para que el supervisor lo reinicie.
5. Sesiones persistentes. Si el worker no debe perder mensajes durante los reinicios, conecta con clean session desactivado y un client ID fijo, y suscríbete con QoS 1. La guía de sesiones persistentes explica el compromiso.

En un broker privado de ConnectMQ, los client ID están vinculados a las credenciales que se te emiten, así que prevé un conjunto de credenciales por cada proceso que se conecte al mismo tiempo.

Elegir QoS, retain y last will desde PHP

Para los comandos enviados desde una app web, QoS 1 es el valor por defecto sensato: el broker confirma la recepción, y el dispositivo debería informar del estado resultante en su propio topic para que la interfaz muestre lo que ocurrió realmente. Evita el flag retain en los comandos. Un on retenido se volvería a enviar al dispositivo cada vez que reconecte, quizá horas después; retain es para el estado, no para las acciones, como explica la guía de mensajes retenidos en MQTT.

El worker suscriptor puede anunciar su propia salud con un last will. Añade ->setLastWillTopic(mqtt_topic_prefix() . '/workers/php/status'), ->setLastWillMessage('offline'), ->setLastWillQualityOfService(1) y ->setRetainLastWill(true) a los ajustes de conexión del worker, y publica un online retenido justo después de conectar. Si el worker se bloquea o pierde la red, el broker publica offline en su nombre, y tu monitorización puede alertar sobre ese topic en lugar de deducirlo del silencio. El plan gratuito Serverless de ConnectMQ admite QoS 0 y 1 y mensajes de hasta 256 KB, lo que cubre holgadamente los payloads de comandos y telemetría de una aplicación PHP.

Notas sobre Laravel y Symfony

El mismo código encaja en los frameworks. En Laravel, envuelve el suscriptor en un comando Artisan (php artisan mqtt:listen) y ejecútalo bajo Supervisor, y publica desde un job en cola en lugar de desde el controlador. En Symfony, un comando de consola cumple el mismo papel. El paquete comunitario php-mqtt/laravel-client añade configuración y una facade sobre la misma biblioteca si prefieres ese estilo.

Solución de problemas

Empieza con ConnectMQ

Dale a tu aplicación PHP un broker privado y seguro en pocos minutos. Crea un broker Serverless gratuito en app.connectmq.com sin tarjeta de crédito, pon las credenciales generadas en tu archivo de entorno y usa el cliente en el navegador del portal para probar tus publicaciones. Un broker MQTT gestionado significa que TLS, el escalado y las actualizaciones del broker dejan de ser tarea de tu equipo de PHP. Para otros lenguajes de backend, consulta el tutorial de cliente MQTT en Ruby y el tutorial de Paho MQTT C.

Preguntas frecuentes

¿php-mqtt/client es compatible con MQTT 5?

No. Es compatible con MQTT 3.1 y 3.1.1, que ConnectMQ acepta junto con MQTT 5.0. Para la mayoría de los flujos de comandos de la web al dispositivo, 3.1.1 es suficiente.

¿Puedo suscribirme a topics MQTT dentro de una petición web en PHP?

Puedes, pero no deberías. Las suscripciones necesitan un proceso que permanezca conectado, así que ejecútalas en un worker CLI y limita las peticiones web a publicaciones breves.

¿Por qué a veces no llega mi publicación QoS 1 desde PHP?

Probablemente el script terminó antes de que el cliente procesara la confirmación del broker. Llama a $mqtt->loop(true, true) después de publicar para que la cola se vacíe antes de disconnect().

¿Cómo conecto PHP a MQTT por TLS?

Activa setUseTls(true) en los ajustes de conexión y conéctate al puerto 8883. La verificación del par funciona de serie con brokers que usan certificados de confianza pública, como broker.connectmq.com.