Otimização de desempenho do cliente MQTT: inflight, batching e buffers

A maioria dos problemas de "MQTT lento" não está no broker. Está em um publicador que espera cada confirmação, em uma janela inflight de 20 sobre um link de 100 ms ou em um callback de assinante que grava no banco de dados antes de retornar. Otimizar o desempenho do cliente MQTT se resume a poucos ajustes: max inflight, a forma de publicar, se você agrupa mensagens, o tamanho das filas, o keep-alive e o próprio socket TCP.

Este guia explica cada parâmetro, mostra como configurá-lo nas bibliotecas mais comuns e entrega um script para medir o efeito antes e depois.

Por que o round-trip time define o teto

Com QoS 1, cada mensagem precisa de um PUBACK. Se o cliente permite apenas N mensagens sem confirmação, o throughput por conexão fica limitado a aproximadamente:

máx. mensagens/s ≈ janela inflight ÷ round-trip time

Com 20 mensagens inflight e um round trip de 50 ms, são cerca de 400 mensagens por segundo, por mais rápido que o broker seja. Aumente a janela para 200 e o teto vai para cerca de 4.000. Em uma LAN com round trip de 1 ms, a janela padrão raramente é o limite; entre regiões ou em rede celular, quase sempre é.

QoS 0 não tem confirmações, então a janela inflight não se aplica — ali os limites são os buffers do socket, a CPU e os limites por cliente do broker. QoS 2 precisa de dois round trips por mensagem e atinge o teto mais cedo. Veja os benchmarks de QoS 0 vs 1 vs 2 para entender o trade-off.

Os parâmetros em resumo

O broker tem a palavra final. No MQTT 5, o Receive Maximum do broker no CONNACK informa ao cliente quantas mensagens QoS 1/2 ele pode ter em trânsito; um cliente em conformidade não pode ultrapassá-lo, seja qual for a configuração local. Se aumentar o ajuste do cliente não muda nada, confira esse valor — controle de fluxo e Receive Maximum explica como lê-lo.

1. Aumente o max inflight — com cuidado

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="testmqtt-tuned-pub")
client.max_inflight_messages_set(100)   # padrão 20
client.max_queued_messages_set(10000)   # limita a memória em vez de ilimitada

MqttConnectOptions opts = new MqttConnectOptions();
opts.setMaxInflight(100);               // padrão 10
opts.setKeepAliveInterval(30);

Uma janela maior não é de graça. Cada mensagem inflight fica na memória do cliente até ser confirmada e, após uma reconexão, todas são reenviadas — uma janela grande transforma uma oscilação de rede em uma rajada de mensagens duplicadas. Aumente aos poucos e pare quando o throughput deixar de melhorar.

2. Publique de forma assíncrona

O maior erro é esperar cada publicação terminar antes de enviar a próxima:

for reading in readings:
    info = client.publish("testmqtt/tuning/data", reading, qos=1)
    info.wait_for_publish()             # serializa tudo no RTT

No paho-mqtt, publish() já enfileira a mensagem e retorna imediatamente. Deixe assim e acompanhe a conclusão em on_publish ou guardando os objetos MQTTMessageInfo para conferir no final.

No MQTT.js 5, publishAsync retorna uma promise. Dar await em cada uma dentro de um loop tem o mesmo problema; disparar todas de uma vez não tem limite algum. Uma janela pequena dá velocidade e back-pressure ao mesmo tempo:

import mqtt from "mqtt";

const client = await mqtt.connectAsync("mqtt://broker.connectmq.com:1883", {
  clientId: `testmqtt-tune-${process.pid}`,
  keepalive: 30,
});

const WINDOW = 100;                      // nosso próprio limite inflight
const N = 5000;
const payload = Buffer.alloc(256, "x");
const pending = new Set();
const started = performance.now();

for (let i = 0; i < N; i++) {
  const p = client.publishAsync("testmqtt/tuning/js", payload, { qos: 1 })
    .finally(() => pending.delete(p));
  pending.add(p);
  if (pending.size >= WINDOW) await Promise.race(pending);
}
await Promise.all(pending);

const secs = (performance.now() - started) / 1000;
console.log(`${N} msgs em ${secs.toFixed(2)} s = ${(N / secs).toFixed(0)} msg/s`);
await client.endAsync();

3. Agrupe mensagens pequenas

Cada mensagem MQTT carrega um cabeçalho fixo, o nome do topic e — em QoS 1 — um PUBACK de 4 bytes de volta. Para uma leitura de 20 bytes, o topic pode ser maior que os dados. O batching envia uma mensagem com muitas leituras:

import json, time

buffer, last_flush = [], time.monotonic()

def add_reading(client, value):
    global last_flush
    buffer.append({"ts": time.time(), "v": value})
    if len(buffer) >= 50 or time.monotonic() - last_flush >= 1.0:
        client.publish("testmqtt/tuning/batch", json.dumps(buffer), qos=1)
        buffer.clear()
        last_flush = time.monotonic()

Esvazie o lote por tamanho ou por tempo, para que um sensor pouco ativo ainda entregue em até um segundo. Os trade-offs são reais: a latência cresce até o intervalo de envio, um lote perdido leva muitas leituras e os consumidores precisam entender arrays. Use a calculadora de tamanho de pacote para ver quanto overhead você realmente economiza e mantenha os lotes bem abaixo do tamanho máximo de pacote do broker. No MQTT 5, os topic aliases são uma alternativa mais leve que encurta nomes de topic longos sem mudar o formato do payload.

4. Dimensione buffers e filas de propósito

Uma fila de saída ilimitada esconde problemas: quando a rede é mais lenta que o produtor, a memória cresce até o processo ser encerrado. Uma fila limitada transforma isso em um erro explícito que você pode tratar. No paho-mqtt, ao atingir max_queued_messages_set(n), o MQTTMessageInfo retornado por publish() vem com rc == MQTT_ERR_QUEUE_SIZE e nada é enfileirado — registre isso e decida se vai descartar, amostrar ou desacelerar.

Do lado receptor, o equivalente é a velocidade de processamento do assinante. Os callbacks rodam na thread de rede ou no event loop do cliente; enquanto seu on_message está ocupado, o cliente não lê o socket, não envia PUBACKs e não responde ao keep-alive. Passe o trabalho para uma fila e retorne imediatamente:

import queue, threading

work = queue.Queue(maxsize=10000)

def on_message(client, userdata, msg):
    work.put_nowait((msg.topic, msg.payload))   # nunca bloquear aqui

def worker():
    while True:
        topic, payload = work.get()
        save_to_database(topic, payload)        # trabalho lento fora da thread de rede

threading.Thread(target=worker, daemon=True).start()

Se a fila encher mesmo assim, o consumidor é de fato lento demais — escale com shared subscriptions ou veja consumidores lentos. No cliente Go do Paho, SetOrderMatters(false) permite que os handlers de mensagem rodem em paralelo em vez de um após o outro, ao custo de perder garantias de ordem entre handlers.

5. Keep-alive e TCP_NODELAY

O keep-alive não deixa nada mais rápido. Ele decide a rapidez com que uma conexão morta é percebida: o broker derruba um cliente após cerca de 1,5 × o intervalo de keep-alive sem tráfego. De 30 a 60 segundos serve para a maioria dos dispositivos conectados; use mais em links com bateria. A maioria das bibliotecas pula o PINGREQ enquanto outros pacotes estão fluindo. Detalhes em MQTT keep alive.

O algoritmo de Nagle pode atrasar mensagens pequenas: o sistema operacional segura um segmento pequeno enquanto há dados anteriores sem confirmação e, combinado com ACKs atrasados do outro lado, isso gera picos de latência de dezenas de milissegundos. Desativá-lo com TCP_NODELAY envia cada pacote imediatamente. Se a sua biblioteca já faz isso depende da biblioteca e da versão; ativar por conta própria não faz mal. No paho-mqtt, faça isso no callback on_socket_open:

import socket

def on_socket_open(client, userdata, sock):
    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

client.on_socket_open = on_socket_open

Meça antes e depois

Otimizar sem medir é chute. Este script publica um número fixo de mensagens QoS 1 com uma janela inflight definida e informa a taxa com que foram confirmadas:

import os, sys, threading, time
import paho.mqtt.client as mqtt

N = 5000
INFLIGHT = int(sys.argv[1]) if len(sys.argv) > 1 else 20
acked, done = 0, threading.Event()

def on_publish(client, userdata, mid, reason_code, properties):
    global acked
    acked += 1
    if acked == N:
        done.set()

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-bench-{os.getpid()}")
client.max_inflight_messages_set(INFLIGHT)
client.on_publish = on_publish
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()
time.sleep(1)

start = time.perf_counter()
for _ in range(N):
    client.publish("testmqtt/tuning/bench", b"x" * 256, qos=1)
done.wait(timeout=120)
secs = time.perf_counter() - start
print(f"inflight={INFLIGHT}: {acked}/{N} confirmadas em {secs:.2f} s = {acked / secs:.0f} msg/s")
client.loop_stop()
client.disconnect()

Rode com python inflight_bench.py 10 e depois com 20, 50, 100, 200. A taxa deve subir e depois se estabilizar; o joelho dessa curva é a janela certa para aquele caminho de rede. Repita a partir de onde seus dispositivos realmente estão — um resultado medido ao lado do broker diz pouco sobre um dispositivo em outro continente. Enquanto o teste roda, abra o cliente online e assine testmqtt/tuning/# para confirmar que as mensagens chegam.

Para testes maiores com muitos clientes, passe para ferramentas dedicadas — o guia de teste de throughput cobre essas opções.

Otimize nesta ordem: pare de esperar cada publicação, aumente a janela inflight até a curva estabilizar, agrupe apenas se o overhead por mensagem ainda dominar, limite suas filas e mantenha os callbacks rápidos. Cada passo é uma ou duas linhas de código, e o script de benchmark mostra se valeu a pena.

Perguntas frequentes