MQTT em PHP com php-mqtt/client

A forma padrão de usar MQTT em PHP é a biblioteca php-mqtt/client: instale com o Composer, crie um MqttClient, chame connect() e depois publish() de qualquer lugar e subscribe() com uma closure dentro de um loop CLI de longa duração. Ela fala MQTT 3.1 e 3.1.1 sobre TCP ou TLS, não precisa de extensões do PHP além das padrão e tem um wrapper oficial para Laravel.

Este tutorial cobre a publicação (incluindo o detalhe do QoS 1 que quase todo mundo deixa passar), um subscriber robusto, TLS com credenciais, Last Will e como testar cada parte.

Instalação

O php-mqtt/client exige PHP 8 e Composer:

composer require php-mqtt/client

Existem duas formas muito diferentes de o PHP conversar com um broker, e confundi-las causa a maioria dos problemas:

Workers do PHP-FPM são reciclados e requisições expiram, então nunca tente manter uma assinatura aberta dentro de uma requisição web.

Publicar uma mensagem

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

Dois detalhes importam:

• O client ID aleatório. Cada requisição é uma conexão separada. Se duas requisições simultâneas usassem o mesmo ID fixo, o broker desconectaria a primeira — veja client ID duplicado. O gerador de client ID mostra bons formatos.
• loop(true, true) depois de publicar com QoS 1 ou 2. publish() apenas enfileira a mensagem; a confirmação é tratada dentro do loop. O segundo argumento faz o loop() retornar assim que nada estiver pendente. Com QoS 0 você pode desconectar direto. Os níveis de QoS do MQTT explicam o trade-off.

Abrir uma conexão por requisição é ok para eventos ocasionais. Se uma página publica a cada acesso, ou você envia rajadas de mensagens, passe o trabalho para um worker de fila que mantém uma conexão aberta — o padrão mostrado em MQTT no Laravel.

Mensagens retidas para estado

O quarto argumento de publish() é a flag retain. Uma mensagem retida fica armazenada no broker e é entregue imediatamente a cada novo subscriber, o que a torna ideal para topics de "estado atual", como a configuração de um dispositivo editada em um painel administrativo em 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);

Use retain para estado, não para eventos: uma mensagem retida de "botão pressionado" seria reenviada a todo cliente que assinasse depois. Mantenha os payloads como JSON compacto com um timestamp, para que os consumidores saibam a idade do estado.

Assinar em um worker CLI

Um subscriber registra uma closure por filtro de topic e então chama loop(true), que bloqueia e despacha mensagens até você interrompê-lo:

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

O curinga + corresponde a um nível de topic, então esse filtro recebe testmqtt/php/sensors/room1, mas não testmqtt/php/sensors/room1/battery. Confira os filtros com o topic matcher antes de implantar.

Rode php subscriber.php em um terminal e php publish.php em outro. Você deve ver a leitura impressa em milissegundos.

Sobreviver a desconexões

Um worker que roda por semanas uma hora vai perder a conexão: reinícios do broker, timeouts de NAT, deploys. Quando isso acontece, a biblioteca lança uma exceção a partir do loop(). O padrão robusto mais simples é um loop externo que reconecta com back-off e assina de novo:

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

O Last Will publica offline (retido) se o worker morrer sem uma desconexão limpa, e o worker o sobrescreve com online após cada conexão — um indicador de saúde barato que outros serviços podem assinar. Last Will no MQTT cobre os detalhes.

Usar uma sessão limpa aqui é proposital: o worker assina de novo a cada conexão, então não depende de o broker guardar nada. Se você precisa que mensagens fiquem enfileiradas enquanto o worker está offline, mude para connect($settings, false) com um client ID fixo e teste contra o seu próprio broker privado — o endpoint compartilhado broker.connectmq.com não guarda sessões offline. Veja sessões persistentes.

O php-mqtt/client também tem setReconnectAutomatically(true) com setMaxReconnectAttempts() e setDelayBetweenReconnectAttempts() (em milissegundos), útil para publishers que mantêm uma conexão aberta entre jobs.

TLS e autenticação

Brokers de produção exigem TLS na porta 8883 e credenciais. Só mudam a porta e as configurações:

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

Sem um arquivo de CA explícito, o OpenSSL do PHP usa o repositório de confiança do sistema (ou openssl.cafile no php.ini). Mantenha a verificação do peer ligada; desativá-la esconde problemas reais de certificado.

Usando no Laravel

Os mesmos autores mantêm o php-mqtt/laravel-client, que lê as configurações de conexão de config/mqtt-client.php e do .env e expõe uma 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 assinar, você chama MQTT::connection() dentro de um comando Artisan e roda loop(true) ali. O padrão completo — jobs em fila para publicar, um comando supervisionado para assinar, e testes — está em MQTT no Laravel.

Erros comuns

Teste

Veja o que o seu código PHP publica com um subscriber no terminal ou com o cliente MQTT online:

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

Depois envie uma mensagem para o seu subscriber, incluindo uma inválida para provar que a validação de 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 fim, mate o worker com kill -9 e confirme que o status retido offline aparece em testmqtt/php/worker/status. Há mais receitas de linha de comando em exemplos de mosquitto_pub e mosquitto_sub.

Veja os topics de PHP no cliente online

O php-mqtt/client torna o MQTT em PHP simples assim que você respeita o modelo de processos do PHP: publique a partir das requisições com client IDs únicos e um loop(true, true) para QoS 1, assine apenas em workers CLI supervisionados que reconectam sozinhos e ative TLS e credenciais antes de ir para produção.

Perguntas frequentes