如何在消费者断开连接时停止阻塞的XMPP监听Socket？

你的核心痛点是阻塞式的sock.recv()会卡住执行流程，导致无法及时响应WebSocket消费者的断开事件。下面是两种可靠的解决方案：

方案1：线程+主动关闭Socket打破阻塞 #

把XMPP监听逻辑放到独立线程中，当消费者断开时，直接关闭XMPP Socket，让recv()抛出异常从而退出循环。

重构XMPP监听代码（线程化） #

import threading
import socket
import ssl
from asgiref.sync import async_to_sync
from django_channels_redis.core import get_channel_layer

# 全局引用，用于控制线程和Socket
xmpp_listener_thread = None
xmpp_socket = None

def start_xmpp_listener():
    """启动XMPP监听线程（单实例）"""
    global xmpp_listener_thread, xmpp_socket
    if xmpp_listener_thread and xmpp_listener_thread.is_alive():
        return
    
    def _listen():
        global xmpp_socket
        channel_layer = get_channel_layer()
        hostname = "your_xmpp_server_host"
        port = 5223  # 根据实际XMPP端口调整
        
        # 初始化Socket
        raw_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        xmpp_socket = ssl.wrap_socket(
            raw_sock, 
            cert_reqs=ssl.CERT_NONE, 
            ssl_version=ssl.PROTOCOL_TLSv1_2
        )
        
        try:
            xmpp_socket.connect((hostname, port))
            xmpp_socket.sendall("<presence/>".encode())
            
            while True:
                try:
                    presence_data = xmpp_socket.recv(4096)
                    if not presence_data:
                        # XMPP服务器主动关闭连接
                        break
                    # 发送到Channel组，注意解码二进制数据
                    async_to_sync(channel_layer.group_send)(
                        "presence_channel",
                        {"type": "presence", "text": presence_data.decode('utf-8')}
                    )
                except (socket.error, ssl.SSLError):
                    # Socket被关闭，退出循环
                    break
        finally:
            # 确保Socket被关闭
            if xmpp_socket:
                try:
                    xmpp_socket.close()
                except Exception:
                    pass
    
    xmpp_listener_thread = threading.Thread(target=_listen, daemon=True)
    xmpp_listener_thread.start()

def stop_xmpp_listener():
    """停止XMPP监听，关闭Socket打破阻塞"""
    global xmpp_socket
    if xmpp_socket:
        try:
            # 先shutdown再close，确保recv立即抛出异常
            xmpp_socket.shutdown(socket.SHUT_RDWR)
            xmpp_socket.close()
        except Exception:
            pass

修改Consumer代码关联断开事件 #

from asgiref.sync import sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer

class PresenceConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        self.group_name = "presence_channel"
        # 加入Channel组
        await self.channel_layer.group_add(self.group_name, self.channel_name)
        # 启动XMPP监听（单实例控制）
        await sync_to_async(start_xmpp_listener)()
        await self.accept()

    async def disconnect(self, close_code):
        # 离开Channel组
        await self.channel_layer.group_discard(self.group_name, self.channel_name)
        # 停止XMPP监听
        await sync_to_async(stop_xmpp_listener)()

    async def presence(self, event):
        # 转发消息到WebSocket
        await self.send_json(event)

方案2：非阻塞Socket+select监听停止信号 #

把Socket设为非阻塞模式，用select同时监听XMPP Socket的可读事件和一个控制管道，这样既能接收XMPP消息，又能及时响应停止命令。

示例代码 #

import os
import select
import socket
import ssl
import threading
from asgiref.sync import async_to_sync
from django_channels_redis.core import get_channel_layer

# 创建控制管道，用于传递停止信号
stop_pipe_read, stop_pipe_write = os.pipe()
xmpp_thread = None

def start_xmpp_listener():
    global xmpp_thread
    if xmpp_thread and xmpp_thread.is_alive():
        return
    
    def _listen():
        channel_layer = get_channel_layer()
        hostname = "your_xmpp_server_host"
        port = 5223
        
        raw_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        xmpp_socket = ssl.wrap_socket(
            raw_sock, 
            cert_reqs=ssl.CERT_NONE, 
            ssl_version=ssl.PROTOCOL_TLSv1_2
        )
        xmpp_socket.setblocking(False)
        
        try:
            # 处理非阻塞连接
            xmpp_socket.connect_ex((hostname, port))
            select.select([], [xmpp_socket], [])
            xmpp_socket.sendall("<presence/>".encode())
            
            while True:
                # 监听XMPP Socket和停止管道
                ready_socks, _, _ = select.select([xmpp_socket, stop_pipe_read], [], [])
                if stop_pipe_read in ready_socks:
                    # 收到停止信号，退出循环
                    os.read(stop_pipe_read, 1)
                    break
                if xmpp_socket in ready_socks:
                    try:
                        presence_data = xmpp_socket.recv(4096)
                        if not presence_data:
                            break
                        async_to_sync(channel_layer.group_send)(
                            "presence_channel",
                            {"type": "presence", "text": presence_data.decode('utf-8')}
                        )
                    except (socket.error, ssl.SSLError):
                        break
        finally:
            xmpp_socket.close()
            os.close(stop_pipe_read)
            os.close(stop_pipe_write)
    
    xmpp_thread = threading.Thread(target=_listen, daemon=True)
    xmpp_thread.start()

def stop_xmpp_listener():
    # 写入停止信号到管道
    os.write(stop_pipe_write, b'1')

Consumer的修改和方案1一致，在disconnect中调用stop_xmpp_listener()即可。

关键注意点 #

• 如果是多消费者场景，要确保XMPP监听是单实例，避免重复连接XMPP服务器。
• 必须把recv()得到的二进制数据解码为字符串，否则send_json会抛出序列化错误。
• 处理Socket操作的异常，避免线程崩溃。

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