python - 将 threading.Timer 与 asyncio 一起使用

Question

我是 python 的 ascynio 功能的新手，我有一个服务器可以处理来自浏览器的 websocket 请求。这是其工作原理的简化版本:

@asyncio.coroutine
def web_client_connected(self, websocket):
    self.web_client_socket = websocket

    while True:
        request = yield from self.web_client_socket.recv()
        json_val = process_request(request)
        yield from self.socket_queue.put(json_val)

@asyncio.coroutine
def push_from_web_client_json_queue(self):
    while True:
        json_val = yield from self.socket_queue.get()
        yield from self.web_client_socket.send(json_val)

您有一个循环来查找来自客户端的 Web 套接字请求。当它得到一个时，它会对其进行处理并将值放入队列中。另一个循环正在该队列中查找值，当它找到一个时，它会将处理过的值发送回网络套接字。非常简单，而且有效。

我现在想做的是引入一个计时器。当请求到来并完成处理时，我不想立即将响应放回队列中，而是想启动一个 1 分钟的计时器。计时器结束后，我想将响应放入队列。

我试过类似的方法:

@asyncio.coroutine
def web_client_connected(self, websocket):
    self.web_client_socket = websocket

    while True:
        request = yield from self.web_client_socket.recv()
        json_val = process_request(request)
        t = threading.Timer(60, self.timer_done, json_val)
        t.start()

@asyncio.coroutine
def timer_done(self, args):
    yield from self.socket_queue.put(args)

但它不起作用。 timer_done 方法永远不会被调用。如果我删除了 @asyncio.coroutine 装饰器和 yield from，那么 timer_done 确实会被调用，但随后会调用 self.socket_queue。 put(args) 不起作用。

我想我在这里误解了一些基本的东西。你是怎么做到的？

Answer 1

代替计时器，使用 asyncio.ensure_future() 和 asyncio.sleep() :

@asyncio.coroutine
def web_client_connected(self, websocket):
    self.web_client_socket = websocket

    while True:
        request = yield from self.web_client_socket.recv()
        json_val = process_request(request)
        asyncio.ensure_future(web_client_timer(json_val))
        yield

@asyncio.coroutine
def web_client_timer(self, json_val):
    yield from asyncio.sleep(60)
    yield from self.socket_queue.put(json_val)

工作示例:

import asyncio


@asyncio.coroutine
def foo():
    print("enter foo")
    timers = []
    for i in range(10):
        print("Start foo", i)
        yield from asyncio.sleep(0.5)
        print("Got foo", i)
        timers.append(asyncio.ensure_future(timer(i)))
        yield
    print("foo waiting")
    # wait for all timers to finish
    yield from asyncio.wait(timers)
    print("exit foo")


@asyncio.coroutine
def timer(i):
    print("Setting timer", i)
    yield from asyncio.sleep(2)
    print("**** Timer", i)


loop = asyncio.get_event_loop()
resp = loop.run_until_complete(foo())
loop.close()