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?
對比其他協定:
| 特性 | XMPP | MQTT | 專屬協定(LINE/WhatsApp) |
|---|---|---|---|
| 開放性 | ✅ 開放標準 | ✅ 開放標準 | ❌ 封閉 |
| 分散式 | ✅ 可自建伺服器 | ❌ 需要 Broker | ❌ 只能用官方伺服器 |
| 可擴展 | ✅ XML 可擴展 | ⚠️ 有限 | ❌ 無法擴展 |
| 線上狀態 | ✅ 內建 | ❌ 需自己實作 | ✅ 內建 |
| 適用場景 | 即時聊天、社交 | IoT、感測器 | 消費者 App |
🏗️ XMPP 架構
分散式設計
💡 比喻:Email 系統
alice@gmail.com 可以寄信給 bob@yahoo.com
不需要都使用 Gmailalice@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 = 電話線(只提供雙向連線,應用層自己定義格式)| 特性 | XMPP | WebSocket |
|---|---|---|
| 協定層級 | 應用層協定 | 傳輸層協定 |
| 訊息格式 | 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 = True2. 批次處理 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,但支援多個 resource5. 快取 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:協定已定義,開發更快
- 比專屬協定:開放、可自建、可互通
🔗 延伸閱讀
- 上一篇:05-2. MQTT 協定
- 下一篇:05-4. WebRTC 基礎
- 官方 XEPs:https://xmpp.org/extensions/