Python SocketIO Server：如何从一个命名空间触发另一命名空间事件？

你的代码失效的核心原因有两个：

1. 直接实例化SecondNameSpace()调用方法，这不是服务器上已注册的命名空间实例，无法触发真实的客户端事件
2. 异步方法trigger_second_action未使用await调用，导致异步任务未被执行

针对不同角色用户分属不同Namespace、需要跨Namespace广播的场景，提供两种可行的实现方式：

方式一：通过Socket.IO服务器实例获取目标命名空间 #

这种方式依赖已初始化的Socket.IO服务器实例，获取注册后的命名空间对象进行操作：

import socketio

# 创建异步Socket.IO服务器
sio = socketio.AsyncServer(async_mode='asgi')

class FirstNamespace(socketio.AsyncNamespace):
    async def on_connect(self, sid):
        pass

    async def on_first_action(self, sid, data):
        # 处理业务逻辑
        result = some_function(data)
        
        # 获取服务器上已注册的SecondNamespace实例
        second_ns = sio.namespace('/second')
        if second_ns:
            # 异步方法必须用await调用
            await second_ns.trigger_second_action(result)

class SecondNamespace(socketio.AsyncNamespace):
    async def on_connect(self, sid):
        # 用户连接时加入目标房间
        await self.enter_room(sid, room='myroom')

    async def trigger_second_action(self, data):
        # 向房间内的客户端广播事件
        await self.emit('my_event', data, room='myroom')

# 将命名空间注册到服务器
sio.register_namespace(FirstNamespace('/first'))
sio.register_namespace(SecondNamespace('/second'))

def some_function(data):
    # 示例业务处理函数
    return {"processed_data": data}

方式二：直接通过服务器跨命名空间发送事件 #

如果不需要封装额外逻辑，可以直接通过服务器实例指定命名空间发送事件，更简洁：

import socketio

sio = socketio.AsyncServer(async_mode='asgi')

class FirstNamespace(socketio.AsyncNamespace):
    async def on_first_action(self, sid, data):
        result = some_function(data)
        # 直接指定命名空间和房间发送事件
        await sio.emit('my_event', result, room='myroom', namespace='/second')

class SecondNamespace(socketio.AsyncNamespace):
    async def on_connect(self, sid):
        await self.enter_room(sid, room='myroom')

sio.register_namespace(FirstNamespace('/first'))
sio.register_namespace(SecondNamespace('/second'))

def some_function(data):
    return {"processed_data": data}

关键注意事项 #

• 必须使用服务器实例获取已注册的命名空间，不能直接实例化命名空间类
• 所有异步操作（emit、调用异步方法）必须添加await关键字
• 确保目标用户已正确加入指定房间myroom，否则广播无法触达

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