Tutorial de Cliente MQTT em PHP com php-mqtt/client

Um cliente MQTT em PHP permite que a sua aplicação web existente envie comandos a dispositivos e reaja aos dados deles sem acrescentar uma nova linguagem à sua stack. Neste tutorial você vai usar o pacote php-mqtt/client para publicar a partir de uma requisição web, rodar um worker assinante de longa duração, proteger ambos com TLS e manter o worker vivo em produção.

Dois trabalhos bem diferentes para um cliente MQTT em PHP

O modelo de requisições do PHP define como você usa o MQTT, então separe os dois casos desde o início:

Tentar assinar dentro de uma requisição web é o erro mais comum. Os workers do PHP-FPM foram feitos para terminar em milissegundos; uma assinatura precisa de um processo que fique conectado por dias. Mantenha os assinantes em um processo CLI supervisionado por systemd, Supervisor ou pelo seu runtime de contêineres, e reserve as requisições web para publicações curtas.

Instalando o php-mqtt/client

O php-mqtt/client é uma implementação em PHP puro do MQTT 3.1 e 3.1.1 com suporte a TLS, mantida no GitHub e instalada via Composer. Ele não precisa de nenhuma extensão PHP além do suporte padrão a sockets e OpenSSL que a maioria dos builds inclui.

composer require php-mqtt/client

O README do pacote é a documentação oficial do php-mqtt/client e lista todas as configurações de conexão. Para comparar com clientes de outras linguagens, veja o guia de bibliotecas cliente MQTT.

Configurações de conexão compartilhadas

Coloque os detalhes da conexão em um só lugar. O exemplo usa o servidor MQTT público (gratuito) em broker.connectmq.com, destinado a testes e protótipos, via TLS na porta 8883 e passa automaticamente para credenciais privadas quando as variáveis de ambiente estão definidas.

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 é imutável: cada setter retorna uma nova instância, e é por isso que o código reatribui $settings. A verificação TLS usa o repositório de CAs do sistema, que já confia no certificado Let's Encrypt servido pelo ConnectMQ.

Publicando a partir de uma requisição web

Este endpoint recebe um POST de formulário e publica um comando para um dispositivo. Com QoS 1, o broker precisa confirmar a mensagem, e o php-mqtt/client só processa essa confirmação dentro do seu loop, então o código roda o loop até a fila de saída ficar vazia 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);   // aguarda até que o PUBACK seja recebido
    $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';
}

Repare na sanitização da entrada em device. Os usuários nunca devem conseguir injetar #, + ou / em um nome de topic, ou poderão endereçar dispositivos que não são deles.

Uma conexão TLS por requisição acrescenta latência, normalmente uma ou duas idas e voltas de handshake. Para um painel administrativo que envia alguns comandos por minuto, tudo bem. Para publicação em alto volume a partir do PHP, coloque o trabalho em uma fila (Redis, uma tabela no banco, filas do Laravel) e deixe um worker de longa duração manter uma única conexão MQTT e esvaziá-la.

Um assinante MQTT em PHP de longa duração

O assinante é um script CLI. Ele assina com um curinga, decodifica o JSON e para de forma limpa ao receber SIGTERM, para que o systemd ou o Docker possam reiniciá-lo sem perder uma mensagem no meio do processamento.

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

        // Substitua pela sua própria persistência, ex.: um insert via PDO ou um model do 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);
}

Execute:

php bin/subscriber.php

Em outro terminal, faça o papel de um 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}'

A extensão pcntl está disponível no PHP CLI no Linux, mas geralmente não no PHP-FPM nem no Windows, o que é mais um motivo para manter o assinante como processo CLI.

Rodando o worker em produção

/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) guarda as credenciais do broker privado:

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

Processos PHP de longa duração têm seus próprios modos de falha. Percorra esta checklist antes de confiar no worker:

1. Memória. Use unset em arrays grandes, evite acumular dados em propriedades estáticas e considere encerrar o processo depois de algumas horas para que o systemd reinicie um processo limpo.
2. Conexões de banco de dados. Conexões PDO expiram enquanto o worker está ocioso; reconecte em caso de falha em vez de manter uma conexão para sempre.
3. Client IDs únicos. O publicador web e o worker precisam de client IDs diferentes. Se dois processos compartilham um, o broker desconecta o mais antigo, e eles ficam derrubando um ao outro em loop.
4. Limites de reconexão. Com setMaxReconnectAttempts(10), o loop acaba lançando uma exceção; deixe-o encerrar com código diferente de zero para que o supervisor o reinicie.
5. Sessões persistentes. Se o worker não pode perder mensagens durante reinícios, conecte com clean session desativado e um client ID fixo, e assine com QoS 1. O guia de sessões persistentes explica os prós e contras.

Em um broker privado ConnectMQ, os client IDs ficam vinculados às credenciais emitidas para você, então planeje um conjunto de credenciais por processo que se conecta ao mesmo tempo.

Escolhendo QoS, retain e last will no PHP

Para comandos enviados a partir de um app web, QoS 1 é o padrão sensato: o broker confirma o recebimento, e o dispositivo deve informar o estado resultante no próprio topic para que a UI mostre o que realmente aconteceu. Evite a flag retain em comandos. Um on retido seria reenviado ao dispositivo toda vez que ele reconectasse, possivelmente horas depois; retain é para estado, não para ações, como explica o guia sobre retained messages no MQTT.

O worker assinante pode anunciar a própria saúde com um last will. Adicione ->setLastWillTopic(mqtt_topic_prefix() . '/workers/php/status'), ->setLastWillMessage('offline'), ->setLastWillQualityOfService(1) e ->setRetainLastWill(true) às configurações de conexão do worker e publique um online retido logo após conectar. Se o worker travar ou perder a rede, o broker publica offline em nome dele, e o seu monitoramento pode alertar com base nesse topic em vez de adivinhar a partir do silêncio. O plano gratuito ConnectMQ Serverless suporta QoS 0 e 1 e mensagens de até 256 KB, o que cobre com folga os payloads de comandos e telemetria de uma aplicação PHP.

Observações sobre Laravel e Symfony

O mesmo código se encaixa nos frameworks. No Laravel, encapsule o assinante em um comando Artisan (php artisan mqtt:listen) e rode-o sob o Supervisor, e publique a partir de um job em fila em vez de fazê-lo no controller. No Symfony, um comando de console cumpre o mesmo papel. O pacote da comunidade php-mqtt/laravel-client acrescenta configuração e uma facade sobre a mesma biblioteca, se você preferir esse estilo.

Solução de problemas

Comece a usar o ConnectMQ

Dê à sua aplicação PHP um broker privado e seguro em poucos minutos. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, coloque as credenciais geradas no seu arquivo de ambiente e use o cliente no navegador do portal para testar as suas publicações. Um broker MQTT gerenciado tira TLS, escalabilidade e atualizações do broker das costas da sua equipe PHP. Para outras linguagens de back-end, veja o tutorial de cliente MQTT em Ruby e o tutorial de Paho MQTT C.

Perguntas frequentes

O php-mqtt/client suporta MQTT 5?

Não. Ele suporta MQTT 3.1 e 3.1.1, que o ConnectMQ aceita junto com o MQTT 5.0. Para a maioria dos fluxos de comandos da web para dispositivos, o 3.1.1 é suficiente.

Posso assinar topics MQTT dentro de uma requisição web PHP?

Pode, mas não deveria. Assinaturas precisam de um processo que fique conectado, então rode-as em um worker CLI e reserve as requisições web para publicações curtas.

Por que a minha publicação QoS 1 a partir do PHP às vezes não chega?

Provavelmente o script terminou antes de o cliente processar a confirmação do broker. Chame $mqtt->loop(true, true) depois de publicar para que a fila se esvazie antes do disconnect().

Como conectar o PHP ao MQTT via TLS?

Ative setUseTls(true) nas configurações de conexão e conecte-se à porta 8883. A verificação do peer funciona de imediato com brokers que têm certificados publicamente confiáveis, como o broker.connectmq.com.