我在 Python 3.4 中使用 Asyncio,我将尝试解释我目前正在做什么以及我(认为)是什么导致了这个问题。
一方面,我有一个带有阻塞操作的 UDP 连接框架,我从这个流中获取数据并创建我以 SSE 格式传递给客户端的 json。这一切都很好。
我遇到的问题是,如果我不做任何事情并且客户端断开连接,我将无法正确处理客户端断开连接,我将开始收到此错误:
WARNING [selector_events:613] socket.send() raised exception.
由于循环仍在运行,我一直在寻找彻底打破循环并触发 .close() 的方法,但我在找到的示例中遇到了问题,并且在线资源并不多。
似乎实际可行的一个示例是尝试从客户端读取一行,如果它是一个空字符串,则意味着客户端已断开连接。
while True:
data = (yield from client_reader.readline())
if not data: #client disconnected
break
但是在大约十条消息之后,发送给客户端的所有消息都停止了,我认为这是因为它挂在“data =(来自client_reader.readline())”上,如果我关闭客户端然后它会正确关闭并“结束”连接”确实被调用。任何想法为什么它可能会挂起?我认为我在这一点上对 Asyncio 有一个很好的处理,但这个让我感到困惑。
注意:location() 和 status() 是我从 UDP 套接字获取信息的两个调用——我已经使用相同的代码成功运行了它们多个小时而没有出现问题——减去客户端断开线路。
clients = {}
def accept_client(client_reader, client_writer):
task = asyncio.Task(handle_client(client_reader, client_writer))
clients[task] = (client_writer)
def client_done(task):
del clients[task]
client_writer.close()
log.info("End Connection")
log.info("New Connection")
task.add_done_callback(client_done)
@asyncio.coroutine
def handle_client(client_reader, client_writer):
data = {'result':{'status':'Connection Ready'}}
yield from postmessage(data,client_writer)
while True:
data = (yield from client_reader.readline())
if not data: #client disconnected
break
data = yield from asyncio.wait_for(location(),
timeout=1.0)
yield from postmessage(data,client_writer)
data = yield from asyncio.wait_for(status(),
timeout=1.0)
yield from postmessage(data,client_writer)
@asyncio.coroutine
def postmessage(data, client_writer):
mimetype=('text/event-stream')
response = ('data: {0}\n\n'.format(data).encode('utf-8'))
client_writer.write(response)
client_writer.drain()
更新:如果我在“来自 client_reader 的收益”上添加超时,当它达到它通常会挂起的程度时,我会收到以下错误。
2014-11-17 03:13:56,214 INFO [try:23] End Connection
2014-11-17 03:13:56,214 ERROR [base_events:912] Task exception was never retrieved
future: <Task finished coro=<handle_client() done, defined at try.py:29> exception=TimeoutError()>
Traceback (most recent call last):
File "/opt/python3.4.2/lib/python3.4/asyncio/tasks.py", line 236, in _step
result = next(coro)
File "try.py", line 35, in handle_client
timeout=1.0))
File "/opt/python3.4.2/lib/python3.4/asyncio/tasks.py", line 375, in wait_for
raise futures.TimeoutError()
concurrent.futures._base.TimeoutError
这是一个示例脚本,显示了运行中的错误 - 只需在 python 3.4.2 中运行它,经过 9 次迭代后,它将挂起从客户端读取。
(脚本已完成,您可以运行它自己查看)
import asyncio
import logging
import json
import time
log = logging.getLogger(__name__)
clients = {}
def accept_client(client_reader, client_writer):
task = asyncio.Task(handle_client(client_reader, client_writer))
clients[task] = (client_writer)
def client_done(task):
del clients[task]
client_writer.close()
log.info("End Connection")
log.info("New Connection")
task.add_done_callback(client_done)
@asyncio.coroutine
def handle_client(client_reader, client_writer):
data = {'result':{'status':'Connection Ready'}}
postmessage(data,client_writer)
count = 0
while True:
data = (yield from asyncio.wait_for(client_reader.readline(),timeout=1.0))
if not data: #client disconnected
break
data = yield from asyncio.wait_for(test1(),timeout=1.0)
yield from postmessage(data,client_writer)
data = yield from asyncio.wait_for(test2(),timeout=1.0)
yield from postmessage(data,client_writer)
@asyncio.coroutine
def postmessage(data, client_writer):
mimetype=('text/event-stream')
response = ('data: {0}\n\n'.format(data).encode('utf-8'))
client_writer.write(response)
client_writer.drain()
@asyncio.coroutine
def test1():
data = {'result':{
'test1':{ }
}
}
data = json.dumps(data)
return data
@asyncio.coroutine
def test2():
data = {'result':{
'test2':{ }
}
}
data = json.dumps(data)
return data
def main():
loop = asyncio.get_event_loop()
f = asyncio.start_server(accept_client, host=None, port=2991)
loop.run_until_complete(f)
loop.run_forever()
if __name__ == '__main__':
log = logging.getLogger("")
formatter = logging.Formatter("%(asctime)s %(levelname)s " +
"[%(module)s:%(lineno)d] %(message)s")
# log the things
log.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(formatter)
log.addHandler(ch)
main()
另一个更新:我发现它死了,因为它从客户端的标题中读取所有行,然后在用完行时超时。我认为,我正在寻找的真正答案是,当您实际上不需要从客户端接收数据(除了初始连接)时,如何检测客户端断开连接。