Como Construir um Sistema de Rastreabilidade na Manufatura com Eventos MQTT

Um sistema de rastreabilidade na manufatura responde a duas perguntas com rapidez e evidências: o que entrou neste produto e como ele foi feito, e quais outros produtos compartilham o mesmo material ou as mesmas condições suspeitas. Este guia mostra como projetar eventos de rastreabilidade, publicá-los via MQTT a partir de cada estação, armazená-los em um banco de dados e consultá-los quando chegar uma reclamação de cliente ou um recall de fornecedor.

O que a rastreabilidade realmente exige

A rastreabilidade muitas vezes é reduzida a "imprimimos um número de série". Um sistema útil registra muito mais:

1. Identidade. Um número de série único para cada unidade, ou um ID de lote ou batelada onde o rastreamento por unidade não é viável.
2. Genealogia. Quais números de série de componentes ou lotes de material foram consumidos em qual unidade. É isso que permite o rastreamento para trás e para a frente.
3. Roteiro. Por quais estações a unidade passou, quando e em qual linha.
4. Dados de processo. Os parâmetros em vigor enquanto a unidade era processada: torque e ângulo de um parafuso, temperatura e tempo de uma cura, pressão de uma montagem por interferência.
5. Resultados. Resultados de testes e inspeções, valores medidos e histórico de retrabalho.

Com esses cinco registros você rastreia para trás (de uma unidade com falha até seus materiais, máquinas e parâmetros) e para a frente (de um lote de material suspeito ou de uma janela de tempo ruim até todas as unidades afetadas). O rastreamento para a frente é o que transforma um recall de "tudo o que foi feito em março" em uma lista precisa.

Por que a rastreabilidade baseada em eventos funciona

A rastreabilidade tradicional costuma ser construída fazendo cada estação gravar diretamente em um banco de dados central ou no MES por meio de uma interface sob medida. Isso funciona até você adicionar uma estação, trocar de fornecedor ou precisar dos dados em outro lugar. Uma abordagem baseada em eventos desacopla os produtores dos consumidores: cada estação publica o que aconteceu como uma pequena mensagem, e qualquer número de sistemas assina.

O MQTT se encaixa bem nisso. Estações e gateways de borda publicam por um protocolo leve, o QoS 1 garante que cada evento seja entregue pelo menos uma vez, e novos consumidores, como um dashboard de qualidade ou um serviço de CEP em tempo real, podem ser adicionados sem mexer nas estações. Se o padrão é novo para você, leia primeiro o que faz um broker MQTT.

Uma regra importa acima de tudo: o broker transporta os eventos, o banco de dados os armazena. Um broker MQTT não é um arquivo, então o banco de dados de rastreabilidade é um assinante que persiste cada evento.

Projetando os eventos

Mantenha o vocabulário de eventos pequeno e explícito. Estes seis tipos cobrem a maior parte da manufatura discreta:

Todo evento carrega um event_id (um UUID gerado na origem) e um timestamp de um relógio sincronizado. O ID do evento torna o armazenamento idempotente: o QoS 1 pode ocasionalmente entregar uma mensagem duas vezes, e o banco de dados simplesmente ignora a duplicata.

Estrutura de topics

Use uma hierarquia no estilo ISA-95 para que os eventos fiquem naturalmente agrupados por planta, área, linha e estação:

acme/plant1/assembly/line2/station30/trace/event

Um assinante que precisa de todos os eventos de rastreabilidade da planta assina acme/plant1/+/+/+/trace/event. O artigo sobre topics MQTT no padrão ISA-95 cobre essa nomenclatura em profundidade, e o portal da ConnectMQ reconhece esses topics e agrupa os dispositivos por nível automaticamente.

Exemplos de payload

{
  "event_id": "8f1c2e7a-5b9d-4c11-9e0f-2a7d61c3b4f0",
  "event": "component_consumed",
  "ts": "2026-10-16T07:52:13.118Z",
  "serial": "PMP-26-0041877",
  "station": "station20",
  "part_number": "MOT-0.75KW",
  "component_serial": "M2609-118422"
}

{
  "event_id": "c3a90b5e-0d2f-47b8-8d6e-91f4a7e02b1c",
  "event": "station_result",
  "ts": "2026-10-16T07:55:40.902Z",
  "serial": "PMP-26-0041877",
  "station": "station30",
  "result": "pass",
  "values": {"torque_nm": 12.4, "angle_deg": 186.0, "leak_rate_ml_min": 0.8},
  "program": "SCREW-P7",
  "operator": "OP114"
}

Construindo o assinante de rastreabilidade

O serviço em Python abaixo assina todos os eventos de rastreabilidade sob um prefixo, valida-os e os grava no SQLite. Ele guarda o evento completo como JSON, para não perder nada, e extrai os campos mais consultados para colunas indexadas. O SQLite mantém o exemplo autocontido; o mesmo esquema funciona no PostgreSQL em produção. Ele usa paho-mqtt 2.x e o servidor MQTT público de testes da ConnectMQ.

import json
import sqlite3
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a41b06"
db = sqlite3.connect("traceability.db", check_same_thread=False)
db.executescript("""
CREATE TABLE IF NOT EXISTS events (
    event_id TEXT PRIMARY KEY,
    event TEXT NOT NULL,
    ts TEXT NOT NULL,
    serial TEXT NOT NULL,
    station TEXT,
    topic TEXT,
    payload TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS ix_serial ON events(serial);
CREATE INDEX IF NOT EXISTS ix_station_ts ON events(station, ts);
CREATE TABLE IF NOT EXISTS genealogy (
    serial TEXT NOT NULL,
    component TEXT NOT NULL,
    part_number TEXT,
    PRIMARY KEY (serial, component)
);
CREATE INDEX IF NOT EXISTS ix_component ON genealogy(component);
""")

REQUIRED = {"event_id", "event", "ts", "serial"}

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/+/+/trace/event", qos=1)

def on_message(client, userdata, msg):
    try:
        e = json.loads(msg.payload)
    except json.JSONDecodeError:
        print("invalid JSON on", msg.topic)
        return
    if not REQUIRED.issubset(e):
        print("missing fields", e)
        return
    with db:
        db.execute("INSERT OR IGNORE INTO events VALUES (?,?,?,?,?,?,?)",
                   (e["event_id"], e["event"], e["ts"], e["serial"],
                    e.get("station"), msg.topic, json.dumps(e)))
        if e["event"] == "component_consumed":
            component = e.get("component_serial") or e.get("lot")
            db.execute("INSERT OR IGNORE INTO genealogy VALUES (?,?,?)",
                       (e["serial"], component, e.get("part_number")))

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="trace-writer-a41b06",
                     clean_session=False)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Duas configurações são intencionais. clean_session=False com um client ID fixo pede ao broker que mantenha a assinatura e enfileire as mensagens QoS 1 enquanto o gravador está brevemente offline, dentro dos limites de sessão do broker. INSERT OR IGNORE sobre o ID do evento transforma entregas duplicadas em operações sem efeito.

Publique um evento de teste a partir de um terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -i station30-a41b06 -q 1 \
  -t "connectmq/demo/a41b06/line2/station30/trace/event" \
  -m '{"event_id":"c3a90b5e-0d2f-47b8-8d6e-91f4a7e02b1c","event":"station_result","ts":"2026-10-16T07:55:40Z","serial":"PMP-26-0041877","station":"station30","result":"pass"}'

Consultando os dados de rastreabilidade

O valor do sistema aparece quando alguém faz uma pergunta difícil. Estas três consultas cobrem a maioria das investigações.

Rastreamento para trás: tudo o que aconteceu com uma unidade.

SELECT ts, station, event, payload FROM events
WHERE serial = 'PMP-26-0041877' ORDER BY ts;

Rastreamento para a frente: todas as unidades que consumiram um lote de componente suspeito.

SELECT serial FROM genealogy WHERE component = 'LOT-SEAL-2609-07';

Rastreamento por janela de tempo: todas as unidades processadas em uma estação enquanto ela estava fora de controle.

SELECT DISTINCT serial FROM events
WHERE station = 'station30'
  AND ts BETWEEN '2026-10-16T07:30:00Z' AND '2026-10-16T08:15:00Z';

Combine as consultas para a frente e por janela de tempo com os eventos de expedição e você poderá dizer ao atendimento ao cliente exatamente quais paletes segurar.

Checklist de implementação

1. Defina o ponto de serialização. Decida onde os números de série são criados e faça essa estação publicar unit_created.
2. Escaneie todo componente rastreável. A genealogia só é tão completa quanto a disciplina de leitura na montagem.
3. Sincronize os relógios. Use NTP em todo gateway e PC de estação; as consultas por janela de tempo dependem disso.
4. Gere os IDs de evento na origem. Nunca no banco de dados, ou as duplicatas se tornam indistinguíveis.
5. Publique com QoS 1. Use sessões persistentes para o gravador.
6. Armazene em buffer na borda. As estações devem enfileirar os eventos localmente enquanto a rede estiver fora e publicá-los ao reconectar.
7. Valide e monitore. Conte os eventos por estação por hora e alerte quando uma estação ficar em silêncio.
8. Defina a retenção. Combine por quanto tempo os registros devem ser mantidos para o seu setor e seus clientes e planeje o armazenamento do banco de acordo.
9. Proteja o pipeline com credenciais privadas e TLS, seguindo as boas práticas de segurança MQTT.

Conectando a rastreabilidade à qualidade

A rastreabilidade é mais poderosa quando ligada a dados de qualidade ao vivo. Quando o controle de qualidade IoT na manufatura detecta uma falha, o banco de dados de rastreabilidade mostra quais lotes e máquinas anteriores estavam envolvidos. Quando o CEP sinaliza um período fora de controle, a consulta por janela de tempo lista todas as unidades a segregar. O guia de manufatura inteligente mostra como esses sistemas compartilham uma única espinha dorsal de dados, que também pode alimentar um gêmeo digital da produção.

Hospedagem em produção

Os eventos de rastreabilidade são registros de negócio, então devem ficar em um broker privado e autenticado, nunca em um servidor público de testes. Um broker MQTT gerenciado fornece endpoints TLS e credenciais emitidas para cada estação sem que você precise operar infraestrutura de broker. Plantas com requisitos de isolamento mais rígidos podem usar um cluster Dedicated ou rodar o broker na própria conta de nuvem com BYOC, mantendo-o gerenciado.

Construa sua espinha dorsal de rastreabilidade na ConnectMQ

A ConnectMQ oferece MQTT 3.1.1 e 5.0 com TLS, sessões persistentes, QoS 1 e 2 nos planos pagos, detecção de topics ISA-95 e um explorador de topics ao vivo para verificar os eventos de cada estação. Crie um broker no plano gratuito Serverless em app.connectmq.com, sem cartão de crédito, e conecte hoje mesmo a sua primeira estação.

Perguntas frequentes

O que é um sistema de rastreabilidade na manufatura?

É um sistema que registra a identidade, os componentes, o roteiro, os parâmetros de processo e os resultados de cada unidade ou lote, para que você possa rastrear tanto para trás, a partir de um produto, quanto para a frente, a partir de um material ou evento.

A rastreabilidade deve ser por número de série ou por lote?

Use números de série por unidade quando os produtos forem de alto valor ou relevantes para a segurança, e IDs de lote ou batelada quando a marcação por unidade não for viável. Muitas plantas combinam os dois, com unidades serializadas consumindo materiais rastreados por lote.

O broker MQTT pode armazenar o histórico de rastreabilidade?

Não. O broker transporta os eventos. Um assinante precisa gravar cada evento em um banco de dados, que é o sistema de registro.

Como evitar registros duplicados com QoS 1?

Gere um ID de evento único na origem e torne a inserção no banco de dados idempotente sobre esse ID, para que mensagens reentregues sejam ignoradas.