如何在asyncio.create_server中正确使用contextvars保存客户端会话状态

1. 如何在asyncio.create_server场景下正确保存每个客户端的独立状态？ #

首先明确：asyncio.create_server 的 protocol 参数要求是可调用的工厂对象，每次新客户端连接时，都会调用这个工厂生成一个全新的 Protocol 实例——如果之前遇到同一个实例被共用的情况，大概率是直接传了 Protocol 类本身（而非工厂逻辑），或者错误复用了同一个实例对象。

以下是几种可靠的实现方式：

方法一：利用Protocol实例属性存储状态 #

直接在自定义Protocol类中定义实例属性，每个客户端连接对应独立实例，天然隔离状态：

import asyncio

class StatefulEchoProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        # 每个客户端专属的回复计数状态
        self.reply_count = 0
        print(f"新客户端连接：{transport.get_extra_info('peername')}")

    def data_received(self, data):
        self.reply_count += 1
        message = data.decode()
        response = f"第 {self.reply_count} 次回复：{message}"
        self.transport.write(response.encode())

async def main():
    # 传入类作为工厂（类本身是可调用对象，调用后生成新实例）
    server = await asyncio.create_server(
        StatefulEchoProtocol, '127.0.0.1', 8888
    )
    async with server:
        await server.serve_forever()

asyncio.run(main())

每个客户端连接都会生成独立的 StatefulEchoProtocol 实例，self.reply_count 完全属于当前客户端，不会和其他连接冲突。

方法二：通过Transport映射状态 #

如果无法修改Protocol类，可以用全局容器以客户端的transport为键存储状态：

import asyncio

# 存储每个客户端的状态，key为transport对象
client_states = {}

class EchoProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        # 为新客户端初始化状态
        client_states[transport] = {'reply_count': 0}
        print(f"新客户端连接：{transport.get_extra_info('peername')}")

    def data_received(self, data):
        state = client_states[self.transport]
        state['reply_count'] += 1
        message = data.decode()
        response = f"第 {state['reply_count']} 次回复：{message}"
        self.transport.write(response.encode())

    def connection_lost(self, exc):
        # 客户端断开后清理状态，避免内存泄漏
        del client_states[self.transport]

async def main():
    server = await asyncio.create_server(
        EchoProtocol, '127.0.0.1', 8888
    )
    async with server:
        await server.serve_forever()

asyncio.run(main())

关于contextvars的错误修复 #

之前出现LookupError，是因为没有在客户端对应的上下文里初始化ContextVar。Protocol的回调（如data_received）运行在事件循环的回调上下文，需要在connection_made中为当前客户端设置变量值：

import asyncio
from contextvars import ContextVar

reply_count_var = ContextVar('reply_count', default=0)

class ContextEchoProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        # 在当前客户端的上下文初始化状态
        reply_count_var.set(0)
        print(f"新客户端连接：{transport.get_extra_info('peername')}")

    def data_received(self, data):
        try:
            current_count = reply_count_var.get()
            new_count = current_count + 1
            reply_count_var.set(new_count)
            message = data.decode()
            response = f"第 {new_count} 次回复：{message}"
            self.transport.write(response.encode())
        except LookupError:
            self.transport.write(b"状态未初始化")

async def main():
    server = await asyncio.create_server(
        ContextEchoProtocol, '127.0.0.1', 8888
    )
    async with server:
        await server.serve_forever()

asyncio.run(main())

不过这种方式没必要，Protocol实例本身已经天然隔离客户端状态，用实例属性更直接。

2. contextvars与直接使用全局变量的区别是什么？ #

• 隔离性差异：全局变量是进程级共享，所有线程、异步任务、客户端连接读写的都是同一个变量，修改会影响所有访问者；contextvars是上下文绑定的，每个异步任务（或线程）拥有独立的上下文副本，修改自身上下文变量不会干扰其他上下文。
• 传递逻辑差异：全局变量需要手动管控访问（比如加锁），contextvars会自动跟随上下文传递——比如异步任务调用子任务时，子任务能自动继承父任务的上下文变量，无需手动传参。
• 适用场景差异：全局变量适合存储进程级共享配置；contextvars适合在异步任务链、回调函数中传递状态，避免层层传参的繁琐，或在多线程/异步场景下隔离不同任务的状态。

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