10-3. XMPP 協定詳解

深入理解 XMPP 的分散式架構、即時訊息與線上狀態管理

💬 XMPP 協定詳解

🎯 什麼是 XMPP?

💡 比喻:去中心化的 Email 系統 + 即時通訊

就像 Email:
- user@gmail.com 可以寄信給 friend@yahoo.com
- 不需要都用同一家服務(分散式)

但又像 LINE/WhatsApp:
- 即時傳送訊息
- 看得到對方在線與否
- 可以群組聊天

XMPP(Extensible Messaging and Presence Protocol) 原名 Jabber,是一種開放、可擴展的即時通訊協定,基於 XML 格式進行訊息傳輸。

為什麼選擇 XMPP?

對比其他協定:

特性XMPPMQTT專屬協定(LINE/WhatsApp)
開放性✅ 開放標準✅ 開放標準❌ 封閉
分散式✅ 可自建伺服器❌ 需要 Broker❌ 只能用官方伺服器
可擴展✅ XML 可擴展⚠️ 有限❌ 無法擴展
線上狀態✅ 內建❌ 需自己實作✅ 內建
適用場景即時聊天、社交IoT、感測器消費者 App

🏗️ XMPP 架構

分散式設計

💡 比喻:Email 系統
alice@gmail.com 可以寄信給 bob@yahoo.com
不需要都使用 Gmail
alice@example.com ──┐
                    ├──> example.com Server ──┐
bob@example.com ────┘                         │
                                              │ Server-to-Server
charlie@other.com ──┐                        │
                    ├──> other.com Server ────┘
dave@other.com ─────┘

JID(Jabber ID)格式

user@domain/resource

範例:
alice@chat.example.com/mobile    # 手機
alice@chat.example.com/desktop   # 電腦
alice@chat.example.com/laptop    # 筆電

同一個使用者可以同時在多個裝置登入!

組成部分:

  • user:使用者名稱(可省略,如系統帳號)
  • domain:伺服器域名(必填)
  • resource:資源標識(區分不同裝置,可省略)

📨 XML Stanzas(基本訊息單元)

XMPP 使用 XML 格式傳輸資料,有三種基本 Stanza:

1️⃣ Message(訊息)

💡 比喻:寄信
寄出去就算了,不管對方有沒有收到(除非對方主動回覆)
<message
    to="bob@example.com"
    from="alice@example.com"
    type="chat">
  <body>Hello, how are you?</body>
</message>

Message Type:

  • chat:一對一聊天
  • groupchat:群組聊天
  • headline:通知(不需要回覆)
  • normal:類似 Email(可以沒有立即回覆)

Python 範例:

import slixmpp

class ChatBot(slixmpp.ClientXMPP):
    def __init__(self, jid, password):
        super().__init__(jid, password)
        self.add_event_handler("session_start", self.start)
        self.add_event_handler("message", self.message)

    async def start(self, event):
        self.send_presence()
        await self.get_roster()

    async def message(self, msg):
        # 只處理聊天訊息
        if msg['type'] in ('chat', 'normal'):
            print(f"收到訊息來自 {msg['from']}: {msg['body']}")

            # 回覆訊息
            msg.reply(f"你說:{msg['body']}").send()

# 使用
bot = ChatBot('bot@example.com', 'password')
bot.connect()
bot.process(forever=True)

2️⃣ Presence(線上狀態)

💡 比喻:LINE 的「在線中」、「離開」狀態
讓好友知道你現在的狀態,不用問「你在嗎?」
<!-- 上線 -->
<presence>
  <show>chat</show>
  <status>I'm available to chat</status>
</presence>

<!-- 離開 -->
<presence>
  <show>away</show>
  <status>In a meeting</status>
</presence>

<!-- 忙碌 -->
<presence>
  <show>dnd</show>  <!-- Do Not Disturb -->
  <status>Focusing on work</status>
</presence>

<!-- 下線 -->
<presence type="unavailable"/>

Presence Show 值:

  • chat:有空聊天 💬
  • away:暫時離開 ⏸️
  • xa:Extended Away(長時間離開)🌙
  • dnd:請勿打擾 🔕
  • (無 show):在線 ✅

Python 範例:

class PresenceBot(slixmpp.ClientXMPP):
    async def start(self, event):
        # 發送線上狀態
        self.send_presence(pshow='chat', pstatus='有空聊天')

        # 設定離開狀態
        # self.send_presence(pshow='away', pstatus='開會中')

        # 設定請勿打擾
        # self.send_presence(pshow='dnd', pstatus='專注工作')

        await self.get_roster()

    def __init__(self, jid, password):
        super().__init__(jid, password)
        self.add_event_handler("session_start", self.start)
        self.add_event_handler("changed_status", self.on_presence_changed)

    async def on_presence_changed(self, presence):
        # 監聽好友狀態變化
        print(f"{presence['from']} 的狀態:{presence['show']} - {presence['status']}")

bot = PresenceBot('user@example.com', 'password')
bot.connect()
bot.process(forever=True)

3️⃣ IQ(Info/Query)

💡 比喻:打電話詢問(必須要有回應)
「請問你的電話號碼是?」→ 「是 0912-345-678」

IQ 是 請求-回應 模式,用於查詢或設定資料。

<!-- 請求:查詢 Roster(聯絡人列表) -->
<iq type="get" id="roster1">
  <query xmlns="jabber:iq:roster"/>
</iq>

<!-- 回應:Roster 資料 -->
<iq type="result" id="roster1">
  <query xmlns="jabber:iq:roster">
    <item jid="bob@example.com" name="Bob" subscription="both"/>
    <item jid="charlie@example.com" name="Charlie" subscription="both"/>
  </query>
</iq>

IQ Type:

  • get:查詢資料
  • set:設定資料
  • result:成功回應
  • error:錯誤回應

Python 範例:

async def get_roster(self):
    # 查詢 Roster
    roster = await self.get_roster()

    print("=== 聯絡人列表 ===")
    for jid in roster:
        name = roster[jid]['name'] or jid
        subscription = roster[jid]['subscription']
        print(f"{name} ({jid}): {subscription}")

👥 Roster(聯絡人管理)

💡 比喻:通訊錄
儲存你的好友列表,就像手機的聯絡人

Subscription(訂閱關係)

Alice 想加 Bob 為好友:

1. Alice → Bob: 訂閱請求
   <presence type="subscribe" to="bob@example.com"/>

2. Bob → Server: 接受請求
   <presence type="subscribed" to="alice@example.com"/>

3. Bob → Alice: 反向訂閱(互加好友)
   <presence type="subscribe" to="alice@example.com"/>

4. Alice → Bob: 接受
   <presence type="subscribed" to="bob@example.com"/>

結果:雙向訂閱(subscription="both")

Subscription 狀態:

  • none:無關係
  • to:我訂閱對方(我能看到對方狀態)
  • from:對方訂閱我(對方能看到我的狀態)
  • both:互相訂閱(互為好友)

Python 實作:

class RosterManager(slixmpp.ClientXMPP):
    async def start(self, event):
        self.send_presence()
        await self.get_roster()

    def add_friend(self, jid, name=None):
        """新增好友"""
        # 發送訂閱請求
        self.send_presence_subscription(pto=jid)

        # 更新 Roster
        self.update_roster(jid, name=name)

        print(f"已發送好友請求給 {jid}")

    def remove_friend(self, jid):
        """移除好友"""
        # 取消訂閱
        self.send_presence_subscription(pto=jid, ptype='unsubscribe')

        # 從 Roster 移除
        self.del_roster_item(jid)

        print(f"已移除好友 {jid}")

    def __init__(self, jid, password):
        super().__init__(jid, password)
        self.add_event_handler("session_start", self.start)
        self.add_event_handler("presence_subscribe", self.on_subscribe_request)

    async def on_subscribe_request(self, presence):
        """處理好友請求"""
        from_jid = presence['from']
        print(f"收到好友請求來自 {from_jid}")

        # 自動接受(實際應用中應該詢問使用者)
        self.send_presence_subscription(pto=from_jid, ptype='subscribed')

        # 反向訂閱
        self.send_presence_subscription(pto=from_jid, ptype='subscribe')

# 使用
manager = RosterManager('alice@example.com', 'password')
manager.connect()
manager.process(block=False)

# 新增好友
manager.add_friend('bob@example.com', name='Bob')

🎪 MUC(Multi-User Chat)群組聊天

💡 比喻:LINE 群組
多人可以在同一個聊天室裡對話,每個人有不同的角色和權限

加入群組

<!-- 加入聊天室 -->
<presence to="room@conference.example.com/Alice">
  <x xmlns="http://jabber.org/protocol/muc"/>
</presence>

發送群組訊息

<!-- 發送給所有人 -->
<message to="room@conference.example.com" type="groupchat">
  <body>Hello everyone!</body>
</message>

Python 實作:

class GroupChatBot(slixmpp.ClientXMPP):
    def __init__(self, jid, password, room, nick):
        super().__init__(jid, password)
        self.room = room
        self.nick = nick

        # 註冊 MUC 插件
        self.register_plugin('xep_0045')  # Multi-User Chat

        self.add_event_handler("session_start", self.start)
        self.add_event_handler("groupchat_message", self.on_groupchat_message)
        self.add_event_handler("muc::%s::got_online" % self.room, self.on_user_join)
        self.add_event_handler("muc::%s::got_offline" % self.room, self.on_user_leave)

    async def start(self, event):
        self.send_presence()
        await self.get_roster()

        # 加入聊天室
        self.plugin['xep_0045'].join_muc(self.room, self.nick)

        print(f"已加入聊天室:{self.room} (暱稱:{self.nick})")

    async def on_groupchat_message(self, msg):
        # 處理群組訊息
        if msg['mucnick'] != self.nick:  # 忽略自己的訊息
            print(f"[{msg['mucnick']}] {msg['body']}")

            # 回應特定關鍵字
            if 'hello' in msg['body'].lower():
                self.send_message(
                    mto=self.room,
                    mbody=f"Hi {msg['mucnick']}!",
                    mtype='groupchat'
                )

    async def on_user_join(self, presence):
        print(f"✅ {presence['from'].resource} 加入聊天室")

    async def on_user_leave(self, presence):
        print(f"❌ {presence['from'].resource} 離開聊天室")

    def send_group_message(self, message):
        """發送群組訊息"""
        self.send_message(
            mto=self.room,
            mbody=message,
            mtype='groupchat'
        )

# 使用
bot = GroupChatBot(
    jid='bot@example.com',
    password='password',
    room='myroom@conference.example.com',
    nick='ChatBot'
)

bot.connect()
bot.process(block=False)

# 發送訊息
bot.send_group_message("Hello everyone!")

🔐 XMPP 安全性

1️⃣ TLS 加密

<!-- 客戶端請求 TLS -->
<starttls xmlns="urn:ietf:params:xml:ns:xmpp-tls"/>

<!-- 伺服器回應 -->
<proceed xmlns="urn:ietf:params:xml:ns:xmpp-tls"/>

<!-- 升級為 TLS 連線 -->

Python 設定:

class SecureBot(slixmpp.ClientXMPP):
    def __init__(self, jid, password):
        super().__init__(jid, password)

        # 強制使用 TLS
        self.use_tls = True
        self.use_ssl = False

        # 或使用 SSL(直接加密,通常用 5223 port)
        # self.use_ssl = True

bot = SecureBot('user@example.com', 'password')
bot.connect()

2️⃣ SASL 身份驗證

SASL(Simple Authentication and Security Layer)提供多種認證機制:

  • PLAIN:明文密碼(需搭配 TLS)
  • SCRAM-SHA-1:加密挑戰-回應
  • DIGEST-MD5:MD5 雜湊(已過時)
  • EXTERNAL:使用 TLS 憑證認證
<!-- 客戶端選擇認證機制 -->
<auth xmlns="urn:ietf:params:xml:ns:xmpp-sasl" mechanism="SCRAM-SHA-1">
  base64(client-first-message)
</auth>

<!-- 伺服器挑戰 -->
<challenge xmlns="urn:ietf:params:xml:ns:xmpp-sasl">
  base64(server-first-message)
</challenge>

<!-- 客戶端回應 -->
<response xmlns="urn:ietf:params:xml:ns:xmpp-sasl">
  base64(client-final-message)
</response>

<!-- 認證成功 -->
<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>

3️⃣ 端對端加密(OTR / OMEMO)

💡 比喻:密碼鎖箱
訊息在你的裝置上就加密,伺服器也無法解密
只有對方的裝置能解開

OMEMO(推薦):

# 使用 OMEMO 插件
self.register_plugin('xep_0384')  # OMEMO Encryption

# 發送加密訊息
self.plugin['xep_0384'].encrypt_message(
    msg,
    mto='recipient@example.com'
)

🔧 完整實作:即時聊天系統

聊天客戶端:

import slixmpp
import asyncio
from datetime import datetime

class ChatClient(slixmpp.ClientXMPP):
    def __init__(self, jid, password):
        super().__init__(jid, password)

        # 註冊插件
        self.register_plugin('xep_0030')  # Service Discovery
        self.register_plugin('xep_0199')  # XMPP Ping

        # 註冊事件處理器
        self.add_event_handler("session_start", self.on_session_start)
        self.add_event_handler("message", self.on_message)
        self.add_event_handler("changed_status", self.on_presence_changed)

        self.messages = []  # 訊息歷史

    async def on_session_start(self, event):
        """連線成功"""
        print(f"✅ 登入成功:{self.boundjid}")

        # 發送線上狀態
        self.send_presence(pshow='chat', pstatus='有空聊天')

        # 獲取聯絡人列表
        await self.get_roster()

        # 列出所有好友
        print("\n=== 好友列表 ===")
        roster = self.client_roster
        for jid in roster:
            presence = roster.presence(jid)
            if presence:
                for resource, pres in presence.items():
                    status = pres.get('show', 'available')
                    print(f"{jid}/{resource}: {status}")
            else:
                print(f"{jid}: offline")

    async def on_message(self, msg):
        """收到訊息"""
        if msg['type'] in ('chat', 'normal'):
            from_jid = msg['from'].bare
            body = msg['body']
            timestamp = datetime.now().strftime('%H:%M:%S')

            # 儲存訊息
            self.messages.append({
                'from': from_jid,
                'body': body,
                'timestamp': timestamp
            })

            print(f"\n[{timestamp}] {from_jid}:")
            print(f"  {body}")

    async def on_presence_changed(self, presence):
        """好友狀態變化"""
        jid = presence['from'].bare
        show = presence.get('show', 'available')
        status = presence.get('status', '')

        if presence['type'] == 'unavailable':
            print(f"❌ {jid} 離線")
        else:
            print(f"✅ {jid} 上線 ({show}) - {status}")

    def send_chat_message(self, to_jid, message):
        """發送訊息"""
        self.send_message(
            mto=to_jid,
            mbody=message,
            mtype='chat'
        )
        print(f"已發送給 {to_jid}: {message}")

    def set_status(self, show, status):
        """設定狀態"""
        self.send_presence(pshow=show, pstatus=status)
        print(f"狀態已更新:{show} - {status}")


# 命令列介面
async def main():
    import sys

    if len(sys.argv) < 3:
        print("使用方法: python xmpp_client.py user@example.com password")
        sys.exit(1)

    jid = sys.argv[1]
    password = sys.argv[2]

    # 建立客戶端
    client = ChatClient(jid, password)

    # 連線
    client.connect()
    client.process(block=False)

    # 互動式命令列
    print("\n指令:")
    print("  /msg <JID> <訊息>  - 發送訊息")
    print("  /status <show> <狀態>  - 設定狀態 (chat/away/dnd)")
    print("  /quit  - 退出")

    while True:
        try:
            cmd = input("\n> ")

            if cmd.startswith('/msg '):
                parts = cmd.split(' ', 2)
                if len(parts) == 3:
                    _, to_jid, message = parts
                    client.send_chat_message(to_jid, message)
                else:
                    print("格式錯誤")

            elif cmd.startswith('/status '):
                parts = cmd.split(' ', 2)
                if len(parts) == 3:
                    _, show, status = parts
                    client.set_status(show, status)
                else:
                    print("格式錯誤")

            elif cmd == '/quit':
                print("再見!")
                client.disconnect()
                break

        except KeyboardInterrupt:
            print("\n再見!")
            client.disconnect()
            break

        await asyncio.sleep(0.1)

if __name__ == '__main__':
    asyncio.run(main())

使用方法:

# 啟動客戶端
python xmpp_client.py alice@example.com password123

# 發送訊息
> /msg bob@example.com Hello!

# 設定狀態
> /status away 開會中

# 退出
> /quit

🎓 常見面試題

Q1:XMPP 和 WebSocket 有什麼不同?

答案:

比喻:
XMPP = 完整的郵政系統(有標準信封格式、郵局規則)
WebSocket = 電話線(只提供雙向連線,應用層自己定義格式)
特性XMPPWebSocket
協定層級應用層協定傳輸層協定
訊息格式XML(已定義)自訂(JSON/Binary)
功能完整 IM 功能只提供雙向通訊
線上狀態✅ 內建❌ 需自己實作
好友管理✅ Roster❌ 需自己實作
分散式✅ 支援⚠️ 需自己實作
頻寬較高(XML)較低(二進位)

何時用 XMPP:

  • 需要完整的即時通訊功能
  • 需要分散式架構(自建伺服器)
  • 需要與其他 XMPP 系統互通

何時用 WebSocket:

  • 需要自訂協定
  • 需要極低延遲
  • 需要傳輸二進位資料(圖片、影片)

Q2:XMPP 的 Presence 如何運作?

答案:

Presence 廣播機制:

1. Alice 上線:
   Alice → Server: <presence/>
   Server → 所有訂閱者(Bob, Charlie): Alice is online

2. Alice 更新狀態:
   Alice → Server: <presence><show>away</show></presence>
   Server → 所有訂閱者: Alice is away

3. Alice 下線:
   Alice → Server: <presence type="unavailable"/>
   Server → 所有訂閱者: Alice is offline

優點:

  • 訂閱者自動收到狀態更新,不需要輪詢
  • 伺服器統一管理,不會遺漏

實作範例:

def on_changed_status(self, presence):
    jid = presence['from'].bare
    show = presence.get('show', 'available')

    if presence['type'] == 'unavailable':
        print(f"{jid} 離線了")
        # 更新 UI:顯示為灰色
    elif show == 'dnd':
        print(f"{jid} 請勿打擾")
        # 更新 UI:顯示紅點
    elif show == 'away':
        print(f"{jid} 離開")
        # 更新 UI:顯示黃點
    else:
        print(f"{jid} 在線")
        # 更新 UI:顯示綠點

省電技巧(行動裝置):

# 設定較長的 Presence 更新間隔
self.schedule('presence_update', 300, self.send_presence, repeat=True)
# 每 5 分鐘更新一次,而不是每次狀態變化都更新

Q3:如何實作 XMPP 訊息已讀回執?

答案:

使用 XEP-0184: Message Delivery Receipts 擴展。

流程:

<!-- 1. Alice 發送訊息,請求回執 -->
<message to="bob@example.com" id="msg001">
  <body>Hello!</body>
  <request xmlns="urn:xmpp:receipts"/>
</message>

<!-- 2. Bob 收到訊息,發送回執 -->
<message to="alice@example.com">
  <received xmlns="urn:xmpp:receipts" id="msg001"/>
</message>

<!-- 3. Alice 收到回執,顯示「已讀」 -->

Python 實作:

class ReceiptBot(slixmpp.ClientXMPP):
    def __init__(self, jid, password):
        super().__init__(jid, password)

        # 註冊回執插件
        self.register_plugin('xep_0184')  # Message Delivery Receipts

        self.add_event_handler("message", self.on_message)
        self.add_event_handler("receipt_received", self.on_receipt)

    async def on_message(self, msg):
        if msg['type'] in ('chat', 'normal'):
            print(f"收到訊息:{msg['body']}")

            # 檢查是否請求回執
            if msg['request_receipt']:
                # 發送回執
                msg.reply()
                msg['receipt'] = msg['id']
                msg.send()

    async def on_receipt(self, msg):
        """收到回執"""
        msg_id = msg['receipt']
        print(f"✓✓ 訊息 {msg_id} 已讀")
        # 更新 UI:顯示雙勾

# 發送訊息時請求回執
def send_with_receipt(self, to_jid, body):
    msg = self.make_message(mto=to_jid, mbody=body, mtype='chat')
    msg['request_receipt'] = True  # 請求回執
    msg.send()

顯示狀態:

Alice 的訊息列表:
  Hello!  ✓   (已送達)
  How are you?  ✓✓  (已讀)
  What's up?  ⏳  (發送中)

Q4:XMPP 如何處理離線訊息?

答案:

XEP-0160: Best Practices for Handling Offline Messages

當使用者離線時,伺服器會儲存訊息,等使用者上線時一次性推送。

流程:

1. Alice 離線
2. Bob 發送訊息給 Alice
   Bob → Server: <message to="alice@example.com">Hello</message>
3. Server 儲存訊息(離線訊息佇列)
4. Alice 上線
   Alice → Server: <presence/>
5. Server 推送所有離線訊息
   Server → Alice: <message from="bob@example.com" delay="...">Hello</message>

延遲標記(XEP-0203):

<message from="bob@example.com" to="alice@example.com">
  <body>Hello</body>
  <delay xmlns="urn:xmpp:delay" stamp="2025-01-15T10:30:00Z"/>
</message>

Python 處理:

async def on_message(self, msg):
    # 檢查是否為離線訊息
    if msg['delay']['stamp']:
        sent_time = msg['delay']['stamp']
        print(f"📦 離線訊息(發送時間:{sent_time})")
        print(f"   {msg['from']}: {msg['body']}")
    else:
        print(f"💬 即時訊息")
        print(f"   {msg['from']}: {msg['body']}")

伺服器設定(Prosody):

-- prosody.cfg.lua
modules_enabled = {
    "offline";  -- 啟用離線訊息
}

-- 離線訊息儲存上限
offline_max_messages = 100;

Q5:如何優化 XMPP 的效能?

答案:

多種優化策略:

1. 使用 Stream Compression(XEP-0138):

<!-- 客戶端請求壓縮 -->
<compress xmlns="http://jabber.org/protocol/compress">
  <method>zlib</method>
</compress>

<!-- 伺服器接受 -->
<compressed xmlns="http://jabber.org/protocol/compress"/>
<!-- 之後的資料都壓縮傳輸 -->
# Python 啟用壓縮
self.register_plugin('xep_0138')  # Stream Compression
self.use_compression = True

2. 批次處理 Presence:

# 不好:每次狀態變化都廣播
def update_status_frequently():
    while True:
        self.send_presence()
        time.sleep(10)  # 每 10 秒更新

# 好:延遲批次更新
presence_queue = []

def queue_presence_update(show, status):
    presence_queue.append((show, status))

async def batch_send_presence():
    while True:
        await asyncio.sleep(60)  # 每分鐘批次處理
        if presence_queue:
            latest = presence_queue[-1]  # 只發送最新狀態
            self.send_presence(pshow=latest[0], pstatus=latest[1])
            presence_queue.clear()

3. 使用 Binary XML(EXI):

EXI (Efficient XML Interchange)
將 XML 編碼為二進位格式,大幅減少資料量

<message to="bob@example.com">
  <body>Hello</body>
</message>

原始 XML:67 bytes
EXI 編碼:15 bytes(省 78%)

4. 連線池(多資源共用):

# 同一使用者的多個裝置共用連線
client_desktop = XMPP('alice@example.com/desktop', 'password')
client_mobile = XMPP('alice@example.com/mobile', 'password')

# Server 只維護一個 session,但支援多個 resource

5. 快取 Roster:

import pickle

# 儲存 Roster 到本地
async def cache_roster(self):
    roster = await self.get_roster()
    with open('roster_cache.pkl', 'wb') as f:
        pickle.dump(roster, f)

# 啟動時載入快取
def load_roster_cache(self):
    try:
        with open('roster_cache.pkl', 'rb') as f:
            return pickle.load(f)
    except:
        return None

# 減少啟動時的網路請求

📝 總結

XMPP 是一個功能完整的即時通訊協定:

  • 開放:任何人都能架設伺服器 🏠
  • 分散式:不依賴單一服務提供商 🌍
  • 可擴展:XML 格式易於擴展 🔧
  • 成熟:已有 20+ 年歷史,生態系完整 📚

記憶口訣:

  • 「開(開放)、分(分散)、擴(可擴展)、熟(成熟)」

適用場景:

  • 💬 企業即時通訊(Slack、Discord 的開源替代)
  • 🎮 遊戲聊天系統
  • 📞 VoIP 訊號控制(配合 Jingle)
  • 🏢 內部溝通系統

XMPP vs 其他協定:

  • 比 MQTT:功能更完整(Presence、Roster)
  • 比 WebSocket:協定已定義,開發更快
  • 比專屬協定:開放、可自建、可互通

🔗 延伸閱讀

0%