Python中使用JWT Token建立STOMP连接（CLI对接RabbitMQ）

核心思路 #

GUI中使用的wss://:JWT_TOKEN@URL/[PATH]本质是利用空用户名+JWT作为密码的Basic认证方式，Python的stomp.py库不会自动解析URL中的这种认证格式，需要手动配置认证参数并指定WebSocket传输方式。

步骤实现 #

1. 安装依赖

确保安装所需的库：

pip install stomp.py websocket-client

2. 完整连接代码

import stomp
from stomp.transport import WebSocketTransport

JWT_TOKEN = "your-actual-jwt-token-here"
WSS_URL = "wss://your-domain.com/your-stomp-path"  # 对应GUI中的URL部分，不含JWT
QUEUE_NAME = "/queue/your-target-queue"  # 要订阅的队列/主题

class MyListener(stomp.ConnectionListener):
    def on_message(self, headers, body):
        print(f"收到消息: {body}")

    def on_error(self, headers, message):
        print(f"错误: {message}")

    def on_disconnected(self):
        print("连接断开")

# 配置WebSocket传输，启用SSL
transport = WebSocketTransport(
    urls=[WSS_URL],
    use_ssl=True,
)

# 初始化连接：用户名为空字符串，密码为JWT Token
conn = stomp.Connection(transport=transport)
conn.set_listener("", MyListener())

# 开启调试模式（可选，用于排查问题）
conn.set_debug(True)

# 连接并订阅队列
conn.connect(username="", passcode=JWT_TOKEN, wait=True)
conn.subscribe(destination=QUEUE_NAME, id=1, ack="auto")

# 保持连接运行（CLI程序中可根据需求调整，比如用循环等待）
try:
    while True:
        input("按回车键退出...\n")
        break
finally:
    conn.disconnect()

关键注意事项 #

• 认证方式：必须将用户名设为空字符串，密码传入JWT Token，stomp.py会自动生成对应的Basic Auth请求头（base64编码的:JWT_TOKEN），和GUI的连接方式一致。
• WebSocket传输：必须显式使用WebSocketTransport，否则stomp.py会默认尝试TCP连接，而非WebSocket。
• 路径正确性：确保WSS_URL中的路径和Nginx配置的STOMP路由一致（比如RabbitMQ默认的WebSocket STOMP路径是/ws/stomp）。
• SSL配置：因为是wss协议，use_ssl必须设为True。

故障排查 #

• 开启调试模式（conn.set_debug(True)），查看STOMP帧的交互日志，确认认证头是否正确发送。
• 验证JWT Token的有效性：可以用工具解码Token，检查过期时间和权限。
• 检查RabbitMQ插件：确保rabbitmq_web_stomp插件已启用（执行rabbitmq-plugins enable rabbitmq_web_stomp）。
• 检查Nginx配置：确认路由已正确转发到RabbitMQ的web-stomp端口（默认15674），且没有额外的头校验拦截请求。

内容的提问来源于stack exchange，提问作者AVarf