MQTT no Node.js

O MQTT no Node.js é uma combinação muito usada em IoT, telemetria, automação residencial, dispositivos móveis e comunicação em redes instáveis. O protocolo segue o modelo publish/subscribe: clientes publicam mensagens em tópicos e outros clientes recebem somente os tópicos assinados.

MQTT foi projetado para conexões leves e ambientes com largura de banda limitada. Ele oferece diferentes níveis de qualidade de serviço, sessões persistentes, mensagens retidas, Last Will, reconexão e propriedades avançadas no MQTT 5. Mesmo assim, confiabilidade não acontece automaticamente. O projeto precisa escolher QoS, client IDs, retenção, segurança e política de reconexão de acordo com o comportamento esperado.

Neste guia, você aprenderá a usar MQTT.js em Node.js, conectar com TLS, publicar e assinar tópicos, trabalhar com QoS, retained messages, sessões, Last Will, MQTT 5, backpressure e observabilidade.

O que é MQTT?

MQTT é um protocolo de mensageria publish/subscribe. Um broker central recebe publicações e distribui para clientes inscritos. O site oficial da especificação está em MQTT.org.

No ecossistema JavaScript, uma biblioteca popular é MQTT.js, cliente para Node.js e navegador com suporte a TCP, TLS e WebSockets.

Instalação

npm install mqtt

Depois conecte ao broker:

import mqtt from 'mqtt';

const client = mqtt.connect('mqtt://localhost:1883', {
  clientId: `orders-api-${process.pid}`,
  clean: true,
  reconnectPeriod: 1000,
  connectTimeout: 30_000
});

Em produção, evite IDs aleatórios quando precisa de sessão persistente. O broker identifica a sessão pelo clientId.

Eventos principais

client.on('connect', connack => {
  console.log('Conectado', connack.sessionPresent);
});

client.on('reconnect', () => {
  console.log('Reconectando');
});

client.on('offline', () => {
  console.log('Cliente offline');
});

client.on('error', error => {
  console.error('Erro MQTT', error);
});

client.on('close', () => {
  console.log('Conexão encerrada');
});

Não ignore o evento error. Registre o tipo de falha sem expor credenciais.

Assinando um tópico

client.on('connect', async () => {
  const grants = await client.subscribeAsync('devices/+/telemetry', {
    qos: 1
  });

  console.log(grants);
});

O caractere + representa um nível. O curinga # representa vários níveis:

devices/#

Evite assinaturas amplas sem necessidade. Elas aumentam volume e podem expor dados de outros clientes.

Recebendo mensagens

client.on('message', async (topic, payload, packet) => {
  try {
    const data = JSON.parse(payload.toString('utf8'));
    await handleTelemetry(topic, data, packet);
  } catch (error) {
    logger.error({ topic, error }, 'Falha ao processar mensagem MQTT');
  }
});

O payload chega como Buffer. Limite tamanho e valide o objeto após o parse. Formato válido não significa dado confiável.

Publicando

await client.publishAsync(
  'devices/device-42/commands',
  JSON.stringify({
    command: 'restart',
    requestedAt: new Date().toISOString()
  }),
  {
    qos: 1,
    retain: false
  }
);

Use publishAsync para um fluxo baseado em Promise. Em QoS 1 ou 2, a Promise termina após o ciclo de confirmação correspondente.

QoS 0

QoS 0 oferece entrega no máximo uma vez. Não existe confirmação do broker para a publicação. É adequado quando:

• telemetria é frequente;
• perder uma amostra é aceitável;
• latência e baixo overhead são prioritários;
• o próximo valor substitui o anterior.

Não use QoS 0 para comandos financeiros ou eventos que não podem desaparecer.

QoS 1

QoS 1 oferece entrega pelo menos uma vez. Duplicatas são possíveis. O consumidor precisa ser idempotente:

const key = `${deviceId}:${messageId}`;

if (await dedupe.exists(key)) {
  return;
}

await processCommand(command);
await dedupe.save(key);

Veja Idempotência em APIs Node.js.

QoS 2

QoS 2 executa um handshake maior para garantir entrega exatamente uma vez no nível do protocolo. Isso aumenta tráfego, estado e latência. Ainda assim, a operação de negócio pode ser executada duas vezes se a aplicação falhar fora do ciclo MQTT.

Use QoS 2 apenas quando o custo adicional for justificado e teste comportamento durante quedas.

Retained messages

Uma mensagem retida fica armazenada no broker e é entregue ao novo assinante:

await client.publishAsync(
  'devices/device-42/status',
  JSON.stringify({ online: true }),
  { qos: 1, retain: true }
);

Retained é útil para o estado mais recente, não para histórico. Para limpar:

await client.publishAsync(
  'devices/device-42/status',
  Buffer.alloc(0),
  { qos: 1, retain: true }
);

Evite reter comandos, pois um dispositivo que conecta depois pode executar uma ordem antiga.

Last Will

O Last Will é publicado pelo broker quando o cliente desconecta de forma inesperada:

const client = mqtt.connect('mqtts://broker.example.com:8883', {
  clientId: 'device-42',
  will: {
    topic: 'devices/device-42/status',
    payload: JSON.stringify({ online: false }),
    qos: 1,
    retain: true
  }
});

Após conectar, publique o estado online retido. Assim, consumidores conhecem o último estado.

Sessão limpa

Com clean: true, o cliente começa uma nova sessão. Para uma sessão persistente no MQTT 3.1.1:

const client = mqtt.connect(url, {
  clientId: 'billing-consumer',
  clean: false,
  reconnectPeriod: 2000
});

O broker pode manter assinaturas e mensagens QoS pendentes. O clientId precisa permanecer estável.

MQTT 5 e expiração de sessão

const client = mqtt.connect(url, {
  protocolVersion: 5,
  clean: false,
  properties: {
    sessionExpiryInterval: 3600,
    receiveMaximum: 50,
    maximumPacketSize: 1_048_576
  }
});

MQTT 5 adiciona propriedades como expiração, reason codes, response topics, correlation data e user properties.

Expiração de mensagens

await client.publishAsync(topic, payload, {
  qos: 1,
  properties: {
    messageExpiryInterval: 60,
    contentType: 'application/json',
    payloadFormatIndicator: true
  }
});

Comandos temporários não devem ser executados depois de perder relevância.

Request/response

MQTT 5 permite definir um tópico de resposta:

const correlationData = crypto.randomBytes(16);

await client.publishAsync('devices/device-42/request', payload, {
  qos: 1,
  properties: {
    responseTopic: 'services/api/responses',
    correlationData
  }
});

Implemente timeout e limpeza da promessa pendente. Não permita crescimento ilimitado de requests aguardando resposta.

Reconexão

MQTT.js possui reconexão automática:

{
  reconnectPeriod: 2000,
  connectTimeout: 30_000,
  resubscribe: true
}

Adicione jitter quando muitos clientes podem reconectar juntos. Um reinício do broker pode causar tempestade de conexão.

Backoff com jitter

Quando precisa controlar manualmente:

function delayFor(attempt) {
  const base = Math.min(30_000, 500 * 2 ** attempt);
  return Math.floor(base * (0.5 + Math.random()));
}

Consulte Retry com Backoff no Node.js.

TLS

import { readFileSync } from 'node:fs';

const client = mqtt.connect('mqtts://broker.example.com:8883', {
  ca: readFileSync('./certs/ca.pem'),
  cert: readFileSync('./certs/client.crt'),
  key: readFileSync('./certs/client.key'),
  rejectUnauthorized: true
});

Nunca use rejectUnauthorized: false em produção. Veja mTLS no Node.js.

Autenticação

MQTT suporta username e password, certificados e mecanismos avançados. Não grave credenciais no código:

const client = mqtt.connect(url, {
  username: process.env.MQTT_USERNAME,
  password: Buffer.from(process.env.MQTT_PASSWORD)
});

Consulte Gestão de Segredos no Node.js.

ACLs por tópico

O broker deve aplicar autorização:

• dispositivo publica apenas sua telemetria;
• dispositivo recebe apenas seus comandos;
• serviço lê somente tópicos necessários;
• ninguém assina # sem justificativa.

Não confie na aplicação cliente para respeitar fronteiras.

Tópicos

Uma convenção possível:

tenant/{tenantId}/devices/{deviceId}/telemetry
tenant/{tenantId}/devices/{deviceId}/status
tenant/{tenantId}/devices/{deviceId}/commands

Evite dados pessoais e caracteres difíceis. Defina versionamento quando o significado mudar.

Shared subscriptions

Brokers compatíveis permitem distribuir mensagens entre consumidores:

$share/billing/devices/+/telemetry

Apenas uma instância do grupo recebe cada mensagem. Teste ordering e reentrega conforme o broker.

Backpressure

Se o consumidor processa mais devagar que o broker, filas internas crescem. MQTT.js oferece handleMessage para processamento com controle:

client.handleMessage = async (packet, done) => {
  try {
    await processPacket(packet);
    done();
  } catch (error) {
    done(error);
  }
};

Sempre chame done. Caso contrário, o cliente pode travar.

Payload binário

MQTT não exige JSON. Você pode usar MessagePack, Protobuf ou Avro:

await client.publishAsync(topic, encodedBuffer, {
  qos: 1,
  properties: {
    contentType: 'application/x-protobuf'
  }
});

Veja Protobuf no Node.js.

Documentação com AsyncAPI

Documente tópicos, operações e payloads em AsyncAPI. Consulte AsyncAPI no Node.js.

Observabilidade

Meça:

• estado conectado;
• tentativas de reconexão;
• mensagens publicadas e recebidas;
• latência de processamento;
• erros por tópico;
• mensagens duplicadas;
• fila local;
• QoS e reason codes.

Shutdown gracioso

process.once('SIGTERM', async () => {
  try {
    await client.endAsync(false, {
      reasonCode: 0
    });
  } finally {
    process.exit(0);
  }
});

Veja Graceful Shutdown no Node.js.

Testes

Suba um broker descartável com Docker ou Testcontainers e teste:

• publicação e assinatura;
• QoS 1 com duplicata;
• retained message;
• Last Will;
• reconexão;
• sessão persistente;
• ACLs;
• payload inválido.

Erros comuns

• Client ID aleatório com sessão persistente: sessão nunca é retomada.
• Retain em comandos: ordem antiga é executada.
• QoS 1 sem idempotência: efeitos duplicados.
• Assinatura com #: excesso de dados e risco de acesso.
• Sem TLS: credenciais e payload ficam expostos.
• Reconexão sem jitter: tempestade de clientes.
• Payload sem limite: consumo de memória.
• JSON sem validação: contrato informal.

Conclusão

O MQTT no Node.js oferece comunicação leve e resiliente para IoT e sistemas em tempo real. MQTT.js simplifica conexão, publicação, assinatura, reconexão, QoS e recursos do MQTT 5.

Escolha QoS conforme o negócio, trate duplicatas, use TLS e ACLs, limite payloads e documente tópicos. Com sessões, Last Will, expiração e observabilidade bem configurados, MQTT deixa de ser apenas um canal simples e se torna uma base confiável para comunicação distribuída.