Giao Thức Truyền Thông IoT: MQTT, CoAP, AMQP & WebSocket

# docker-compose.yml
services:
  mosquitto:
    image: eclipse-mosquitto:2
    container_name: mqtt-broker
    ports:
      - "1883:1883"   # MQTT
      - "9001:9001"   # WebSocket (cho browser clients)
    volumes:
      - ./mosquitto/config:/mosquitto/config
      - ./mosquitto/data:/mosquitto/data
      - ./mosquitto/log:/mosquitto/log
    restart: unless-stopped

# mosquitto/config/mosquitto.conf
listener 1883
protocol mqtt
 
listener 9001
protocol websockets
 
# Xác thực (chỉ tắt cho development)
allow_anonymous true
 
# Persistence
persistence true
persistence_location /mosquitto/data/
 
# Logging
log_dest file /mosquitto/log/mosquitto.log
log_type all

# Khởi động broker
docker compose up -d
 
# Xác nhận đang chạy
docker compose logs mosquitto
# Output: mosquitto version 2.x.x running

# Quy ước đặt tên topic: {tổ chức}/{vị trí}/{loại thiết bị}/{id thiết bị}/{loại dữ liệu}
factory/building-a/sensors/temp-042/readings
factory/building-a/sensors/temp-042/status
factory/building-a/actuators/fan-01/commands
factory/building-b/sensors/humidity-003/readings

import mqtt from "mqtt";
 
const client = mqtt.connect("mqtt://localhost:1883");
 
client.on("connect", () => {
  // Single-level wildcard (+): khớp chính xác MỘT cấp
  client.subscribe("factory/building-a/sensors/+/readings");
  // Khớp: factory/building-a/sensors/temp-042/readings ✅
  // Khớp: factory/building-a/sensors/humidity-003/readings ✅
  // Không khớp: factory/building-b/sensors/temp-042/readings ❌
 
  // Multi-level wildcard (#): khớp MỌI THỨ phía dưới
  client.subscribe("factory/building-a/#");
  // Khớp: factory/building-a/sensors/temp-042/readings ✅
  // Khớp: factory/building-a/actuators/fan-01/commands ✅
  // Khớp: factory/building-a/alerts ✅
  // Không khớp: factory/building-b/anything ❌
});
 
client.on("message", (topic, message) => {
  const segments = topic.split("/");
  const building = segments[1];
  const deviceType = segments[2];
  const deviceId = segments[3];
 
  console.log(`[${building}/${deviceId}] ${message.toString()}`);
});

import mqtt from "mqtt";
 
const client = mqtt.connect("mqtt://localhost:1883", {
  clientId: "sensor-node-042",
  clean: false,         // Session persistent — broker nhớ subscriptions
  reconnectPeriod: 5000,
});
 
client.on("connect", () => {
  // QoS 0: Tọa độ GPS mỗi giây — mất một cái cũng được
  setInterval(() => {
    client.publish("fleet/truck-07/gps", JSON.stringify({
      lat: 37.7749 + Math.random() * 0.01,
      lng: -122.4194 + Math.random() * 0.01,
      speed: 55 + Math.random() * 10,
      ts: Date.now(),
    }), { qos: 0 });
  }, 1000);
 
  // QoS 1: Đọc nhiệt độ mỗi 30s — chúng ta muốn tất cả
  setInterval(() => {
    client.publish("factory/zone-1/temp/s042", JSON.stringify({
      value: 22.5 + Math.random() * 5,
      unit: "celsius",
      ts: Date.now(),
    }), { qos: 1 });
  }, 30000);
 
  // QoS 2: Lệnh actuator — phải thực thi chính xác một lần
  client.publish("factory/zone-1/actuators/valve-01/cmd", JSON.stringify({
    action: "open",
    percentage: 75,
    requestId: crypto.randomUUID(),
  }), { qos: 2 });
});

// Thiết bị publish trạng thái dưới dạng retained message
client.publish("devices/sensor-042/status", JSON.stringify({
  state: "online",
  firmware: "2.1.0",
  ip: "192.168.1.42",
  uptime: 86400,
}), {
  qos: 1,
  retain: true, // Broker lưu trữ — subscribers mới nhận ngay lập tức
});
 
// Dashboard kết nối 5 phút sau ngay lập tức nhận được trạng thái
// Không cần chờ bản cập nhật trạng thái tiếp theo

const client = mqtt.connect("mqtt://localhost:1883", {
  clientId: "sensor-042",
  // Last Will: được broker publish nếu client này ngắt kết nối đột ngột
  will: {
    topic: "devices/sensor-042/status",
    payload: JSON.stringify({
      state: "offline",
      lastSeen: Date.now(),
    }),
    qos: 1,
    retain: true, // Ghi đè retained message "online"
  },
});
 
// Khi kết nối, publish trạng thái "online"
client.on("connect", () => {
  client.publish("devices/sensor-042/status", JSON.stringify({
    state: "online",
    lastSeen: Date.now(),
  }), { qos: 1, retain: true });
});
 
// Nếu thiết bị crash hoặc mất mạng:
// 1. TCP keepalive timeout (mặc định 60s)
// 2. Broker publish LWT message
// 3. Dashboard thấy trạng thái "offline"
// 4. Alert service kích hoạt thông báo

// Các tính năng MQTT v5.0 trong thực tế
const client = mqtt.connect("mqtt://localhost:1883", {
  protocolVersion: 5,
  clientId: "sensor-042",
});
 
client.on("connect", () => {
  // Message expiry: alert hết hạn sau 5 phút
  client.publish("alerts/high-temp", JSON.stringify({
    device: "sensor-042",
    temp: 45.2,
  }), {
    qos: 1,
    properties: {
      messageExpiryInterval: 300, // 5 phút tính bằng giây
    },
  });
 
  // Request/Response: hỏi thiết bị về config
  client.publish("devices/sensor-042/config/get", "", {
    qos: 1,
    properties: {
      responseTopic: "responses/sensor-042/config",
      correlationData: Buffer.from("req-001"),
    },
  });
 
  // Shared subscription: cân bằng tải giữa 3 processing workers
  client.subscribe("$share/processors/telemetry/#");
  // Ba workers subscribe cùng shared group
  // Mỗi message chỉ đi đến MỘT worker — cân bằng tải tự động
});

import { createServer, request } from "coap";
 
// ── CoAP Server (chạy trên cảm biến/thiết bị) ──
const server = createServer();
 
// Resource nhiệt độ
let currentTemp = 23.5;
 
server.on("request", (req, res) => {
  const path = req.url;
 
  if (path === "/temperature" && req.method === "GET") {
    res.setOption("Content-Format", "application/json");
    res.end(JSON.stringify({
      value: currentTemp,
      unit: "celsius",
      ts: Date.now(),
    }));
  }
 
  if (path === "/led" && req.method === "PUT") {
    const payload = JSON.parse(req.payload.toString());
    console.log(`LED đặt thành: ${payload.state}`);
    res.code = "2.04"; // Changed
    res.end();
  }
 
  if (path === "/temperature" && req.method === "GET" && req.headers.Observe === 0) {
    // Observe: push cập nhật khi giá trị thay đổi
    res.setOption("Content-Format", "application/json");
 
    const interval = setInterval(() => {
      currentTemp += (Math.random() - 0.5) * 0.5;
      res.write(JSON.stringify({
        value: parseFloat(currentTemp.toFixed(1)),
        ts: Date.now(),
      }));
    }, 5000);
 
    res.on("finish", () => clearInterval(interval));
  }
});
 
server.listen(5683, () => {
  console.log("CoAP server đang lắng nghe trên port 5683");
});

// ── CoAP Client (chạy trên gateway hoặc cloud) ──
import { request } from "coap";
 
// GET request đơn giản
const req = request({
  hostname: "sensor-042.local",
  port: 5683,
  pathname: "/temperature",
  method: "GET",
  confirmable: true, // CON message — tin cậy
});
 
req.on("response", (res) => {
  const data = JSON.parse(res.payload.toString());
  console.log(`Nhiệt độ: ${data.value}°C`);
});
 
req.end();
 
// PUT command để điều khiển actuator
const cmdReq = request({
  hostname: "sensor-042.local",
  port: 5683,
  pathname: "/led",
  method: "PUT",
  confirmable: true,
});
 
cmdReq.write(JSON.stringify({ state: "on", brightness: 80 }));
cmdReq.end();
 
// Observe: subscribe thay đổi nhiệt độ
const obsReq = request({
  hostname: "sensor-042.local",
  port: 5683,
  pathname: "/temperature",
  method: "GET",
  observe: true,
});
 
obsReq.on("response", (res) => {
  res.on("data", (chunk: Buffer) => {
    const data = JSON.parse(chunk.toString());
    console.log(`[Observe] Nhiệt độ: ${data.value}°C`);
  });
});
 
obsReq.end();

// Block-wise transfer cho payload lớn (ví dụ: firmware metadata)
// Thư viện CoAP xử lý block negotiation tự động
 
const req = request({
  hostname: "sensor-042.local",
  pathname: "/firmware/info",
  method: "GET",
  // Block2 option: yêu cầu blocks 256-byte
});
 
req.on("response", (res) => {
  // Thư viện tự động ghép lại các blocks
  const firmwareInfo = JSON.parse(res.payload.toString());
  console.log(`Firmware: ${firmwareInfo.version}, Kích thước: ${firmwareInfo.size}`);
});
 
req.end();

# docker-compose.yml
services:
  rabbitmq:
    image: rabbitmq:3-management
    container_name: iot-rabbitmq
    ports:
      - "5672:5672"     # AMQP
      - "15672:15672"   # Management UI
      - "1883:1883"     # MQTT plugin (tùy chọn)
    environment:
      RABBITMQ_DEFAULT_USER: iot_admin
      RABBITMQ_DEFAULT_PASS: secure_password
    volumes:
      - ./rabbitmq/enabled_plugins:/etc/rabbitmq/enabled_plugins
      - rabbitmq_data:/var/lib/rabbitmq
    restart: unless-stopped
 
volumes:
  rabbitmq_data:

# rabbitmq/enabled_plugins
[rabbitmq_management, rabbitmq_mqtt, rabbitmq_web_stomp].

import amqplib from "amqplib";
 
// ── Producer: IoT Gateway gửi telemetry ──
async function publishTelemetry() {
  const connection = await amqplib.connect("amqp://iot_admin:secure_password@localhost");
  const channel = await connection.createChannel();
 
  // Khai báo topic exchange cho dữ liệu IoT
  const exchange = "iot.telemetry";
  await channel.assertExchange(exchange, "topic", { durable: true });
 
  // Publish dữ liệu cảm biến với routing key
  const routingKey = "factory.zone1.temperature";
  const message = {
    deviceId: "sensor-042",
    value: 28.5,
    unit: "celsius",
    timestamp: new Date().toISOString(),
  };
 
  channel.publish(
    exchange,
    routingKey,
    Buffer.from(JSON.stringify(message)),
    {
      persistent: true,        // Sống sót qua broker restart
      contentType: "application/json",
      messageId: crypto.randomUUID(),
      timestamp: Date.now(),
      headers: {
        "x-device-type": "temperature-sensor",
        "x-firmware": "2.1.0",
      },
    }
  );
 
  console.log(`Đã publish: ${routingKey} → ${message.value}°C`);
}

// ── Consumer: Dịch vụ xử lý nhiệt độ ──
async function consumeTemperature() {
  const connection = await amqplib.connect("amqp://iot_admin:secure_password@localhost");
  const channel = await connection.createChannel();
 
  const exchange = "iot.telemetry";
  const queue = "temperature-processor";
 
  await channel.assertExchange(exchange, "topic", { durable: true });
  await channel.assertQueue(queue, {
    durable: true,
    deadLetterExchange: "iot.dlx", // Messages thất bại đến đây
    messageTtl: 3600000,           // TTL 1 giờ
  });
 
  // Bind: chỉ nhận messages nhiệt độ từ bất kỳ zone nào
  await channel.bindQueue(queue, exchange, "factory.*.temperature");
 
  // Prefetch: xử lý 10 messages cùng lúc (flow control)
  await channel.prefetch(10);
 
  channel.consume(queue, (msg) => {
    if (!msg) return;
 
    try {
      const data = JSON.parse(msg.content.toString());
      console.log(`Đang xử lý: ${data.deviceId} → ${data.value}°C`);
 
      // Kiểm tra cảnh báo
      if (data.value > 35) {
        // Publish alert đến routing key khác
        channel.publish(
          exchange,
          "factory.alerts.temperature",
          Buffer.from(JSON.stringify({
            type: "highTemperature",
            device: data.deviceId,
            value: data.value,
            threshold: 35,
            timestamp: new Date().toISOString(),
          })),
          { persistent: true }
        );
      }
 
      channel.ack(msg); // Xác nhận xử lý thành công
    } catch (error) {
      // Từ chối và không requeue — gửi đến dead letter queue
      channel.nack(msg, false, false);
    }
  });
 
  console.log("Temperature processor đang lắng nghe...");
}

async function setupDeadLetterQueue() {
  const connection = await amqplib.connect("amqp://localhost");
  const channel = await connection.createChannel();
 
  // Dead letter exchange
  await channel.assertExchange("iot.dlx", "topic", { durable: true });
  await channel.assertQueue("iot.failed-messages", { durable: true });
  await channel.bindQueue("iot.failed-messages", "iot.dlx", "#");
 
  // Queue chính với dead letter routing
  await channel.assertQueue("temperature-processor", {
    durable: true,
    deadLetterExchange: "iot.dlx",
    deadLetterRoutingKey: "failed.temperature",
    messageTtl: 3600000, // Messages cũ hơn 1 giờ → DLQ
    maxLength: 100000,   // Queue tràn → DLQ
  });
 
  // Giám sát dead letter queue
  channel.consume("iot.failed-messages", (msg) => {
    if (!msg) return;
    const data = JSON.parse(msg.content.toString());
    const reason = msg.properties.headers?.["x-death"]?.[0]?.reason;
    console.error(`Dead letter [${reason}]: ${JSON.stringify(data)}`);
    // Ghi vào hệ thống giám sát, kích hoạt điều tra
    channel.ack(msg);
  });
}

import { WebSocketServer, WebSocket } from "ws";
import mqtt from "mqtt";
 
// Bridge MQTT telemetry đến WebSocket clients (dashboards)
const wss = new WebSocketServer({ port: 8080 });
const mqttClient = mqtt.connect("mqtt://localhost:1883");
 
// Theo dõi topics mà mỗi WebSocket client đang xem
const clientSubscriptions = new Map<WebSocket, Set<string>>();
 
mqttClient.on("connect", () => {
  mqttClient.subscribe("factory/#");
  console.log("MQTT bridge đã kết nối");
});
 
mqttClient.on("message", (topic, payload) => {
  const message = JSON.stringify({
    topic,
    data: JSON.parse(payload.toString()),
    receivedAt: new Date().toISOString(),
  });
 
  // Chuyển tiếp đến WebSocket clients quan tâm
  wss.clients.forEach((ws) => {
    if (ws.readyState !== WebSocket.OPEN) return;
 
    const subs = clientSubscriptions.get(ws);
    if (!subs) return;
 
    // Kiểm tra xem có subscription pattern nào khớp topic này
    for (const pattern of subs) {
      if (topicMatches(pattern, topic)) {
        ws.send(message);
        break;
      }
    }
  });
});
 
wss.on("connection", (ws) => {
  clientSubscriptions.set(ws, new Set());
 
  ws.on("message", (raw) => {
    const msg = JSON.parse(raw.toString());
 
    if (msg.type === "subscribe") {
      clientSubscriptions.get(ws)?.add(msg.topic);
      console.log(`Client đã subscribe: ${msg.topic}`);
    }
 
    if (msg.type === "unsubscribe") {
      clientSubscriptions.get(ws)?.delete(msg.topic);
    }
  });
 
  ws.on("close", () => {
    clientSubscriptions.delete(ws);
  });
});
 
// Khớp topic kiểu MQTT đơn giản
function topicMatches(pattern: string, topic: string): boolean {
  const patternParts = pattern.split("/");
  const topicParts = topic.split("/");
 
  for (let i = 0; i < patternParts.length; i++) {
    if (patternParts[i] === "#") return true;
    if (patternParts[i] === "+") continue;
    if (patternParts[i] !== topicParts[i]) return false;
  }
 
  return patternParts.length === topicParts.length;
}

// Phía trình duyệt: kết nối đến MQTT broker qua WebSocket
import mqtt from "mqtt";
 
// Kết nối đến Mosquitto's WebSocket port (9001)
const client = mqtt.connect("ws://localhost:9001");
 
client.on("connect", () => {
  console.log("Đã kết nối MQTT broker qua WebSocket");
 
  // Subscribe dữ liệu nhiệt độ cho zone cụ thể
  client.subscribe("factory/zone-1/sensors/+/readings");
});
 
client.on("message", (topic, message) => {
  const data = JSON.parse(message.toString());
  const deviceId = topic.split("/")[3];
 
  // Cập nhật UI dashboard
  updateChart(deviceId, data.value, data.timestamp);
  updateGauge(deviceId, data.value);
 
  // Kiểm tra cảnh báo
  if (data.value > 35) {
    showAlert(`${deviceId}: ${data.value}°C vượt ngưỡng!`);
  }
});
 
function updateChart(deviceId: string, value: number, timestamp: number) {
  // Push data point đến chart library (Chart.js, D3, v.v.)
  console.log(`Cập nhật biểu đồ: ${deviceId} = ${value}°C lúc ${new Date(timestamp)}`);
}
 
function updateGauge(deviceId: string, value: number) {
  console.log(`Cập nhật gauge: ${deviceId} = ${value}°C`);
}
 
function showAlert(message: string) {
  console.warn(`CẢNH BÁO: ${message}`);
}

import mqtt from "mqtt";
import { createServer as createCoapServer } from "coap";
import amqplib from "amqplib";
 
// ── Gateway IoT đa giao thức ──
class ProtocolBridge {
  private mqttClient: mqtt.MqttClient;
  private amqpChannel: amqplib.Channel | null = null;
 
  constructor() {
    // Outbound: MQTT cho real-time telemetry
    this.mqttClient = mqtt.connect("mqtt://broker.example.com", {
      clientId: "gateway-bridge-01",
    });
  }
 
  async initialize() {
    // Outbound: AMQP cho xử lý tin cậy
    const amqpConn = await amqplib.connect("amqp://localhost");
    this.amqpChannel = await amqpConn.createChannel();
    await this.amqpChannel.assertExchange("iot.ingestion", "topic", {
      durable: true,
    });
 
    // Inbound: CoAP server cho dữ liệu cảm biến nội bộ
    this.startCoapListener();
 
    console.log("Protocol bridge đã khởi tạo");
  }
 
  private startCoapListener() {
    const server = createCoapServer();
 
    server.on("request", (req, res) => {
      if (req.method !== "POST") {
        res.code = "4.05"; // Method Not Allowed
        res.end();
        return;
      }
 
      const data = JSON.parse(req.payload.toString());
      const deviceId = req.url.split("/")[1]; // /sensor-042/telemetry
 
      // Bridge CoAP → MQTT (thời gian thực)
      this.mqttClient.publish(
        `devices/${deviceId}/telemetry`,
        JSON.stringify(data),
        { qos: 1 }
      );
 
      // Bridge CoAP → AMQP (xử lý tin cậy)
      this.amqpChannel?.publish(
        "iot.ingestion",
        `telemetry.${deviceId}`,
        Buffer.from(JSON.stringify({
          ...data,
          gatewayId: "gateway-01",
          bridgedFrom: "coap",
          receivedAt: new Date().toISOString(),
        })),
        { persistent: true }
      );
 
      res.code = "2.01"; // Created
      res.end();
    });
 
    server.listen(5683, () => {
      console.log("CoAP listener trên port 5683");
    });
  }
}
 
const bridge = new ProtocolBridge();
bridge.initialize();

import mqtt from "mqtt";
 
// Bridge MQTT alerts đến HTTP webhooks
const client = mqtt.connect("mqtt://localhost:1883");
 
const webhooks: Record<string, string> = {
  "alerts/temperature": "https://hooks.slack.com/services/xxx",
  "alerts/device-offline": "https://api.pagerduty.com/events",
  "alerts/battery-low": "https://your-api.com/webhooks/battery",
};
 
client.on("connect", () => {
  // Subscribe tất cả alert topics
  client.subscribe("alerts/#");
});
 
client.on("message", async (topic, payload) => {
  const data = JSON.parse(payload.toString());
 
  // Tìm webhook khớp
  for (const [pattern, url] of Object.entries(webhooks)) {
    if (topic.startsWith(pattern.replace("#", ""))) {
      try {
        await fetch(url, {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            source: "iot-bridge",
            topic,
            data,
            timestamp: new Date().toISOString(),
          }),
        });
        console.log(`Đã chuyển tiếp ${topic} → ${url}`);
      } catch (error) {
        console.error(`Webhook thất bại cho ${topic}: ${error}`);
      }
    }
  }
});

// ❌ XẤU: QoS 2 cho telemetry tần suất cao
setInterval(() => {
  client.publish("sensors/gps", JSON.stringify(gpsData), {
    qos: 2, // 4 round-trips mỗi message × 1000 thiết bị × 1 msg/s = broker quá tải
  });
}, 1000);

// ✅ TỐT: Khớp QoS với tầm quan trọng của dữ liệu
// GPS mỗi giây: QoS 0 (mất một cái cũng được)
client.publish("sensors/gps", JSON.stringify(gpsData), { qos: 0 });
 
// Nhiệt độ mỗi 30s: QoS 1 (chúng ta muốn tất cả bản đọc)
client.publish("sensors/temp", JSON.stringify(tempData), { qos: 1 });
 
// Lệnh actuator: QoS 2 (phải thực thi chính xác một lần)
client.publish("actuators/valve/cmd", JSON.stringify(cmd), { qos: 2 });

// ❌ XẤU: Tên topic không nhất quán, phẳng
client.publish("temp_sensor_42", data);
client.publish("TemperatureSensor/42", data);
client.publish("sensor-temperature-42", data);
// Không thể dùng wildcards hiệu quả, không có phân cấp

// ✅ TỐT: Topics phân cấp nhất quán
client.publish("acme/factory-1/sensors/temp-042/readings", data);
client.publish("acme/factory-1/sensors/humidity-003/readings", data);
client.publish("acme/factory-1/actuators/fan-01/commands", cmd);
// Dễ subscribe: "acme/factory-1/sensors/+/readings"
// Dễ giám sát: "acme/factory-1/#"

// ❌ XẤU: MQTT cho thiết bị chạy pin device-to-gateway trên mạng nội bộ
// MQTT yêu cầu kết nối TCP (3-way handshake + keepalive packets)
// Trên pin đồng xu, TCP keepalives tốn pin ngay cả khi idle
const client = mqtt.connect("mqtt://gateway.local");
// Keepalive mỗi 60s = ~1440 TCP packets/ngày dù không có dữ liệu

// ✅ TỐT: CoAP cho thiết bị nội bộ, chạy pin
// UDP: gửi một packet, không overhead kết nối, không keepalives
const req = request({
  hostname: "gateway.local",
  port: 5683,
  pathname: "/telemetry",
  method: "POST",
  confirmable: false, // NON: fire-and-forget (ít overhead hơn nữa)
});
req.write(JSON.stringify(sensorData));
req.end();
// 0 packets giữa các lần đọc — pin sống lâu hơn hàng tháng

// ❌ XẤU: Dashboard hiển thị "unknown" cho đến bản cập nhật trạng thái tiếp theo
client.publish("devices/sensor-042/status", JSON.stringify({
  state: "online",
}));
// Subscriber mới không thấy gì cho đến lần publish tiếp theo

// ✅ TỐT: Retained message — subscribers mới nhận trạng thái ngay lập tức
client.publish("devices/sensor-042/status", JSON.stringify({
  state: "online",
  firmware: "2.1.0",
  uptime: 86400,
}), { retain: true });
// Dashboard kết nối 5 phút sau ngay lập tức hiển thị "online"