MQTT และ IoT

ทำไมต้องรู้

ภารกิจเฝ้าระวังน้ำท่วมอาจมีโดรนสามลำ เซนเซอร์ระดับน้ำสิบจุด สถานีภาคพื้น แดชบอร์ดของศูนย์บัญชาการ และฐานข้อมูล ถ้าให้ทุกอุปกรณ์เชื่อมกันเองแบบจุดต่อจุด จำนวนการเชื่อมต่อจะเพิ่มเร็วมาก และเพิ่มอุปกรณ์ใหม่แต่ละครั้งต้องแก้ทุกเครื่อง MQTT แก้ปัญหานี้ด้วยตัวกลางที่รับและกระจายข้อความ UAT 314 โมดูล 1 แนะนำ MQTT ไว้แล้ว โมดูลนี้จะลงรายละเอียดจนออกแบบและตั้งค่าระบบได้จริง

สถาปัตยกรรม IoT

อินเทอร์เน็ตของสรรพสิ่ง (Internet of Things, IoT) คือระบบที่อุปกรณ์จำนวนมากวัดค่า ส่งข้อมูล และรับคำสั่งผ่านเครือข่าย โดยทั่วไปแบ่งเป็นสี่ชั้น

อุปกรณ์ปลายทางมักมีพลังงานและกำลังประมวลผลจำกัด เกตเวย์ รวบรวมข้อมูลจากอุปกรณ์ที่ใช้วิทยุระยะไกลกำลังต่ำ เช่น LoRaWAN แล้วแปลงส่งต่อเข้าเครือข่าย IP ตัวอย่างในคลังความรู้โดรนคือ ChirpStack ซึ่งเป็นเซิร์ฟเวอร์เครือข่าย LoRaWAN ที่ส่งข้อมูลทุกข้อความของอุปกรณ์ออกเป็น JSON ผ่าน MQTT ใน topic เช่น application/APPLICATION_ID/device/DEV_EUI/event/up

Publish/subscribe และ topic

ใน MQTT ผู้ส่ง (publisher) ส่งข้อความไปยัง topic ที่ตัวกลางชื่อ โบรกเกอร์ (broker) ส่วนผู้รับ (subscriber) บอกโบรกเกอร์ว่าสนใจ topic ใด โบรกเกอร์กระจายข้อความให้ผู้รับที่ตรงเงื่อนไข ผู้ส่งกับผู้รับไม่ต้องรู้จักกัน ไม่ต้องออนไลน์พร้อมกัน และเพิ่มผู้รับใหม่ได้โดยไม่แก้ผู้ส่ง

topic เป็นข้อความหลายระดับคั่นด้วย / คล้ายเส้นทางไฟล์ ผู้รับใช้ wildcard ได้สองแบบ ตามมาตรฐาน MQTT 5.0

• + แทน หนึ่งระดับพอดี เช่น drone/+/battery ตรงกับ drone/3/battery แต่ไม่ตรงกับ drone/3/gps
• # แทน ทุกระดับที่เหลือ ต้องอยู่ท้ายสุดเท่านั้น และตรงกับระดับแม่ด้วย เช่น drone/# ตรงกับทั้ง drone/3/gps/fix และ drone

def topic_matches(topic_filter, topic):
    f_parts, t_parts = topic_filter.split("/"), topic.split("/")
    for i, f in enumerate(f_parts):
        if f == "#":
            return True
        if i >= len(t_parts):
            return False
        if f != "+" and f != t_parts[i]:
            return False
    return len(f_parts) == len(t_parts)


cases = [("drone/+/battery", "drone/3/battery"), ("drone/+/battery", "drone/3/gps"), ("drone/#", "drone/3/gps/fix"),
         ("drone/+", "drone/3/battery"), ("drone/#", "drone"), ("fleet/+/+/alt", "fleet/a/7/alt")]
for f, t in cases:
    print(f"{f:<16} {t:<18} {topic_matches(f, t)}")

drone/+/battery  drone/3/battery    True
drone/+/battery  drone/3/gps        False
drone/#          drone/3/gps/fix    True
drone/+          drone/3/battery    False
drone/#          drone              True
fleet/+/+/alt    fleet/a/7/alt      True

QoS: รับประกันการส่งสามระดับ

MQTT กำหนด คุณภาพบริการ (Quality of Service, QoS) ต่อข้อความ

QoS สูงขึ้นแลกกับข้อความและเวลาที่มากขึ้น ใน QoS 1 ถ้า PUBACK หายระหว่างทาง ผู้ส่งจะส่ง PUBLISH ซ้ำ ผู้รับจึงได้ข้อความเดียวกันสองครั้ง โปรแกรมผู้รับต้องเตรียมรับมือ

received = []


def deliver_qos1(msg_id, payload, puback_lost):
    attempts = 0
    while True:
        attempts += 1
        received.append((msg_id, payload))
        if not puback_lost or attempts > 1:
            return attempts


print("attempts:", deliver_qos1(7, "RTL", puback_lost=True), "received:", received)

seen, actions = set(), []
for msg_id, payload in received:
    if msg_id in seen:
        continue
    seen.add(msg_id)
    actions.append(payload)
print("actions:", actions)

attempts: 2 received: [(7, 'RTL'), (7, 'RTL')]
actions: ['RTL']

ข้อความค้างและ Last Will

• ข้อความค้าง (retained message) ถ้าส่งพร้อมธง RETAIN โบรกเกอร์จะเก็บข้อความล่าสุดของ topic นั้นไว้ แทนที่ข้อความค้างเดิม ผู้รับที่สมัครทีหลังจะได้ค่าล่าสุดทันทีโดยไม่ต้องรอรอบถัดไป เหมาะกับสถานะ เช่น online หรือค่าตั้งของภารกิจ
• Last Will อุปกรณ์ฝากข้อความไว้กับโบรกเกอร์ตอนเชื่อมต่อ ถ้าการเชื่อมต่อขาดแบบผิดปกติ โบรกเกอร์จะส่งข้อความนั้นแทน เช่น drone/1/status = offline ศูนย์บัญชาการจึงรู้ว่าโดรนหลุดการเชื่อมต่อ

class MiniBroker:
    def __init__(self):
        self.subs = []
        self.retained = {}

    def subscribe(self, name, topic_filter):
        self.subs.append((topic_filter, name))
        for topic, payload in self.retained.items():
            if topic_matches(topic_filter, topic):
                print(f"  {name} <- {topic} = {payload} (retained)")

    def publish(self, topic, payload, retain=False):
        if retain:
            self.retained[topic] = payload
        for topic_filter, name in self.subs:
            if topic_matches(topic_filter, topic):
                print(f"  {name} <- {topic} = {payload}")


broker = MiniBroker()
broker.subscribe("gcs", "drone/+/battery")
broker.publish("drone/1/status", "online", retain=True)
broker.publish("drone/1/battery", "76%")
print("dashboard joins late:")
broker.subscribe("dashboard", "drone/#")
print("drone 1 link drops, broker sends its will:")
broker.publish("drone/1/status", "offline", retain=True)

  gcs <- drone/1/battery = 76%
dashboard joins late:
  dashboard <- drone/1/status = online (retained)
drone 1 link drops, broker sends its will:
  dashboard <- drone/1/status = offline

แดชบอร์ดที่เข้ามาทีหลังได้สถานะ online ทันทีเพราะเป็นข้อความค้าง แต่ไม่ได้ค่าแบตเตอรี่ 76% เพราะข้อความนั้นไม่ได้ตั้ง RETAIN โบรกเกอร์จำลองตัวนี้เขียนเพื่อให้เห็นหลักการเท่านั้น

MQTT 5.0 (มาตรฐาน OASIS ปี 2019) เพิ่มความสามารถสำคัญจาก 3.1.1 เช่น reason code ที่บอกเหตุผลเมื่อทำงานไม่สำเร็จ session expiry ที่กำหนดว่าโบรกเกอร์จำการสมัครของผู้รับไว้นานเท่าใด และ message expiry interval ที่ทิ้งข้อความที่เก่าเกินกำหนด ซึ่งช่วยไม่ให้ส่งคำสั่งที่หมดอายุไปแล้ว

ความสดของข้อมูลและคิว

ข้อความส่งถึงไม่ได้แปลว่าข้อมูลยังใช้ได้ อายุข้อมูล คือเวลาตั้งแต่วัดจนถึงตอนใช้ ถ้าข้อความเข้าคิวเร็วกว่าที่ผู้รับประมวลผลได้ ($\lambda >\mu$) คิวจะยาวขึ้นเรื่อย ๆ และข้อความใหม่ต้องรอนานขึ้น สูตรนี้มาจากหน่วยความรู้ MQTT และ Mosquitto ของคลังความรู้โดรน

lam, mu, q0 = 50, 40, 0
for t in (10, 60, 300):
    q = max(0, q0 + (lam - mu) * t)
    print(f"after {t:>3} s  queue {q:>4} messages  a new message waits about {q / mu:.1f} s")

after  10 s  queue  100 messages  a new message waits about 2.5 s
after  60 s  queue  600 messages  a new message waits about 15.0 s
after 300 s  queue 3000 messages  a new message waits about 75.0 s

Mosquitto และความปลอดภัย

Eclipse Mosquitto เป็นโบรกเกอร์ MQTT โอเพนซอร์สที่ใช้แพร่หลาย รุ่นล่าสุดขณะเขียนคือ 2.1.2 (กุมภาพันธ์ 2026) มีเครื่องมือ mosquitto_pub และ mosquitto_sub สำหรับทดสอบ IANA กำหนดพอร์ต 1883 สำหรับ MQTT ข้อความธรรมดา และ 8883 สำหรับ MQTT บน TLS

ตั้งแต่ Mosquitto 2.0 ถ้าไม่ได้กำหนด listener โบรกเกอร์จะรับเฉพาะการเชื่อมต่อจากเครื่องตัวเอง (loopback) และเมื่อกำหนด listener แล้ว allow_anonymous มีค่าเริ่มต้นเป็น false ค่าเริ่มต้นนี้ป้องกันการเปิดโบรกเกอร์สู่เครือข่ายโดยไม่ตั้งใจ การใช้งานจริงควรเพิ่มอีกสามชั้น

1. TLS เข้ารหัสข้อมูลระหว่างทางผ่านพอร์ต 8883 กันการดักฟัง
2. บัญชีแยกต่ออุปกรณ์ โดรนแต่ละลำมีชื่อผู้ใช้และรหัสผ่านหรือใบรับรองของตัวเอง
3. สิทธิ์ราย topic (ACL) โดรน 1 publish ได้เฉพาะ drone/1/# จึงปลอมตัวส่งข้อมูลในนามโดรน 2 ไม่ได้

ตัวอย่างไคลเอนต์ด้วย paho-mqtt รุ่น 2 ซึ่งต้องระบุ CallbackAPIVersion (โค้ดนี้ต้องมีโบรกเกอร์จริง จึงไม่ถูกรันในการตรวจอัตโนมัติ)

import os
import paho.mqtt.client as mqtt


def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", reason_code)
    client.subscribe("drone/+/battery", qos=1)


def on_message(client, userdata, msg):
    print(msg.topic, msg.payload.decode())


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="gcs-01", protocol=mqtt.MQTTv5)
client.username_pw_set("gcs-01", os.environ["MQTT_PASSWORD"])
client.tls_set(ca_certs="ca.crt")
client.will_set("gcs/gcs-01/status", "offline", qos=1, retain=True)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.field.lan", 8883)
client.loop_forever()

รหัสผ่านอ่านจากตัวแปรสภาพแวดล้อม ไม่เขียนลงในโค้ด ตามหลักที่เรียนใน UAT 104 โมดูล 5

ปฏิบัติการประจำโมดูล

ข้อผิดพลาดที่พบบ่อย

สรุป

• IoT แบ่งเป็นอุปกรณ์ เกตเวย์ โบรกเกอร์และคลาวด์ และแอปพลิเคชัน MQTT เชื่อมทุกชั้นด้วย publish/subscribe
• + แทนหนึ่งระดับ # แทนทุกระดับที่เหลือและต้องอยู่ท้ายสุด
• QoS 0 อาจหาย QoS 1 อาจซ้ำ QoS 2 ครั้งเดียวพอดีแต่ช้าที่สุด ข้อความค้างให้ค่าล่าสุดแก่ผู้รับใหม่ และ Last Will แจ้งเมื่ออุปกรณ์หลุด
• ตั้ง Mosquitto ด้วย TLS บัญชีแยก และ ACL และตรวจอายุข้อมูลเสมอ เพราะคิวที่สะสมทำให้ข้อมูลเก่าได้โดยไม่มีข้อความหาย

แบบฝึกตรวจความเข้าใจ

1. site/+/drone/# ตรงกับ site/a/drone/2/gps หรือไม่
2. drone/+ ตรงกับ drone/2/battery หรือไม่ เพราะอะไร
3. คำสั่ง “ลงจอด” ควรใช้ QoS ระดับใด และควรออกแบบคำสั่งอย่างไร
4. ข้อความเข้า 30 ข้อความต่อวินาที ประมวลผลได้ 25 ข้อความต่อวินาที เริ่มจากคิวว่าง หลัง 2 นาทีคิวยาวเท่าใด และข้อความใหม่รอประมาณกี่วินาที
5. ทำไมจึงไม่ควรเปิดโบรกเกอร์ที่พอร์ต 1883 บนเครือข่ายที่คนอื่นเข้าได้