Optimizar el rendimiento del cliente MQTT: inflight, batching y buffers

La mayoría de los problemas de «MQTT lento» no son del broker. Son un publicador que espera cada confirmación, una ventana inflight de 20 sobre un enlace de 100 ms o un callback de suscriptor que escribe en una base de datos antes de retornar. Optimizar el rendimiento del cliente MQTT se reduce a unos pocos ajustes: max inflight, cómo publicas, si agrupas mensajes, el tamaño de las colas, el keep-alive y el propio socket TCP.

Esta guía explica cada parámetro, muestra cómo configurarlo en librerías habituales y te da un script para medir el efecto antes y después.

Por qué el round-trip time marca el techo

Con QoS 1, cada mensaje necesita un PUBACK. Si el cliente solo permite N mensajes sin confirmar, el throughput por conexión queda limitado aproximadamente a:

máx. mensajes/s ≈ ventana inflight ÷ round-trip time

Con 20 mensajes inflight y un round trip de 50 ms, son unos 400 mensajes por segundo, por rápido que sea el broker. Sube la ventana a 200 y el techo pasa a unos 4.000. En una LAN con 1 ms de round trip, la ventana por defecto rara vez es el límite; entre regiones o sobre red móvil, casi siempre lo es.

QoS 0 no tiene confirmaciones, así que la ventana inflight no aplica: ahí los límites son los buffers del socket, la CPU y los límites por cliente del broker. QoS 2 necesita dos round trips por mensaje, así que alcanza el techo antes. Consulta los benchmarks de QoS 0 vs 1 vs 2 para ver el compromiso.

Los parámetros de un vistazo

El broker tiene la última palabra. En MQTT 5, el Receive Maximum del broker en el CONNACK indica al cliente cuántos mensajes QoS 1/2 puede tener en vuelo; un cliente conforme no debe superarlo, configures lo que configures localmente. Si subir el ajuste del cliente no cambia nada, revisa ese valor: control de flujo y Receive Maximum explica cómo leerlo.

1. Sube max inflight, con cuidado

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="testmqtt-tuned-pub")
client.max_inflight_messages_set(100)   # por defecto 20
client.max_queued_messages_set(10000)   # limita la memoria en vez de ilimitada

MqttConnectOptions opts = new MqttConnectOptions();
opts.setMaxInflight(100);               // por defecto 10
opts.setKeepAliveInterval(30);

Una ventana más grande no sale gratis. Cada mensaje inflight se queda en la memoria del cliente hasta que se confirma, y tras una reconexión se reenvían todos: una ventana grande convierte un corte de red breve en una ráfaga de mensajes duplicados. Auméntala paso a paso y para cuando el throughput deje de mejorar.

2. Publica de forma asíncrona

El error más grande es esperar a que termine cada publicación antes de enviar la siguiente:

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

En paho-mqtt, publish() ya encola el mensaje y retorna de inmediato. Déjalo así y controla la finalización en on_publish o guardando los objetos MQTTMessageInfo para comprobarlos al final.

En MQTT.js 5, publishAsync devuelve una promesa. Hacer await de cada una en un bucle tiene el mismo problema; lanzarlas todas a la vez no tiene ningún límite. Una ventana pequeña te da velocidad y back-pressure a la vez:

import mqtt from "mqtt";

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

const WINDOW = 100;                      // nuestro propio límite 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 en ${secs.toFixed(2)} s = ${(N / secs).toFixed(0)} msg/s`);
await client.endAsync();

3. Agrupa los mensajes pequeños

Cada mensaje MQTT lleva una cabecera fija, el nombre del topic y, con QoS 1, un PUBACK de 4 bytes de vuelta. Para una lectura de 20 bytes, el topic puede ocupar más que los datos. El batching envía un solo mensaje con muchas lecturas:

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

Vacía el lote por tamaño o por tiempo, para que un sensor tranquilo siga entregando en menos de un segundo. Los compromisos son reales: la latencia crece hasta el intervalo de vaciado, un lote perdido se lleva muchas lecturas y los consumidores deben entender arrays. Usa la calculadora de tamaño de paquete para ver cuánto overhead ahorras de verdad, y mantén los lotes muy por debajo del tamaño máximo de paquete del broker. Con MQTT 5, los topic aliases son una alternativa más ligera que acorta nombres de topic largos sin cambiar el formato del payload.

4. Dimensiona buffers y colas a propósito

Una cola de salida ilimitada esconde problemas: cuando la red es más lenta que tu productor, la memoria crece hasta que matan el proceso. Una cola acotada lo convierte en un error explícito que puedes tratar. En paho-mqtt, al alcanzar max_queued_messages_set(n), el MQTTMessageInfo que devuelve publish() trae rc == MQTT_ERR_QUEUE_SIZE y no se encola nada: regístralo y decide si descartar, muestrear o frenar.

En el lado receptor, el equivalente es la velocidad de procesamiento del suscriptor. Los callbacks se ejecutan en el hilo de red o el event loop del cliente; mientras tu on_message está ocupado, el cliente no lee del socket, no envía PUBACKs y no responde al keep-alive. Pasa el trabajo a una cola y retorna enseguida:

import queue, threading

work = queue.Queue(maxsize=10000)

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

def worker():
    while True:
        topic, payload = work.get()
        save_to_database(topic, payload)        # trabajo lento fuera del hilo de red

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

Si aun así la cola se llena, el consumidor es realmente demasiado lento: escálalo con shared subscriptions o consulta consumidores lentos. En el cliente Go de Paho, SetOrderMatters(false) permite que los handlers se ejecuten en paralelo en lugar de uno tras otro, a costa de perder garantías de orden entre handlers.

5. Keep-alive y TCP_NODELAY

El keep-alive no acelera nada. Decide lo rápido que se detecta una conexión muerta: el broker desconecta a un cliente tras aproximadamente 1,5 × el intervalo de keep-alive sin tráfico. Entre 30 y 60 segundos va bien para la mayoría de dispositivos conectados; usa más en enlaces con batería. La mayoría de librerías omiten el PINGREQ mientras fluyen otros paquetes. Detalles en MQTT keep alive.

El algoritmo de Nagle puede retrasar los mensajes pequeños: el sistema operativo retiene un segmento pequeño mientras hay datos anteriores sin confirmar y, combinado con los ACK retardados del otro extremo, produce picos de latencia de decenas de milisegundos. Desactivarlo con TCP_NODELAY envía cada paquete de inmediato. Que tu librería ya lo active depende de la librería y la versión; activarlo tú no hace daño. En paho-mqtt puedes hacerlo en el 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

Mide antes y después

Optimizar sin medir es adivinar. Este script publica un número fijo de mensajes QoS 1 con una ventana inflight dada e informa del ritmo al que se confirmaron:

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} confirmados en {secs:.2f} s = {acked / secs:.0f} msg/s")
client.loop_stop()
client.disconnect()

Ejecútalo con python inflight_bench.py 10 y luego con 20, 50, 100, 200. El ritmo debería subir y luego aplanarse; el codo de esa curva es la ventana adecuada para esa ruta de red. Repite desde donde están realmente tus dispositivos: un resultado medido junto al broker dice poco de un dispositivo en otro continente. Mientras se ejecuta, abre el cliente online y suscríbete a testmqtt/tuning/# para confirmar que llegan los mensajes.

Para pruebas más grandes con muchos clientes, pasa a herramientas dedicadas: la guía de pruebas de throughput las cubre.

Optimiza en este orden: deja de esperar cada publicación, sube la ventana inflight hasta que la curva se aplane, agrupa solo si el overhead por mensaje sigue dominando, acota tus colas y mantén los callbacks rápidos. Cada paso son una o dos líneas de código, y el script de benchmark te dice si mereció la pena.

Preguntas frecuentes