Python的asyncio同步工作

Question

我想利用Python的新asyncio库发送异步HTTP请求。我希望在发送每个请求之前等待几个毫秒（timeout变量） - 但是当然 - 将它们全部异步发送，而不是在发送每个请求之后等待响应。

我做类似如下：

@asyncio.coroutine 
def handle_line(self, line, destination): 
    print("Inside! line {} destination {}".format(line, destination)) 
    response = yield from aiohttp.request('POST', destination, data=line, 
           headers=tester.headers) 
    print(response.status) 
    return (yield from response.read()) 

@asyncio.coroutine 
def send_data(self, filename, timeout): 
    destination='foo' 
    logging.log(logging.DEBUG, 'sending_data') 
    with open(filename) as log_file: 
     for line in log_file: 
      try: 
       json_event = json.loads(line) 
      except ValueError as e: 
       print("Error parsing json event") 
      time.sleep(timeout) 
      yield from asyncio.async(self.handle_line(json.dumps(json_event), destination)) 


loop=asyncio.get_event_loop().run_until_complete(send_data('foo.txt', 1)) 

，我获得（通过打印200级响应）是这样的代码是同步运行的输出。我究竟做错了什么？

Answer 1

有几个问题在这里：

1. 你应该使用asyncio.sleep，不time.sleep，因为后者将阻止事件循环。

2. 你不应该使用yield from的asyncio.async(self.handle_line(...))电话后，因为那样会使脚本块，直到self.handle_line协程完成，这意味着你最终不会同时做任何事情;你处理每一行，等待处理完成，然后继续下一行。相反，您应该在不等待的情况下运行所有​​asyncio.async调用，将Task对象返回到列表中，然后使用asyncio.wait等待它们全部完成后，全部启动它们。

把所有在一起：

@asyncio.coroutine 
def handle_line(self, line, destination): 
    print("Inside! line {} destination {}".format(line, destination)) 
    response = yield from aiohttp.request('POST', destination, data=line, 
           headers=tester.headers) 
    print(response.status) 
    return (yield from response.read()) 

@asyncio.coroutine 
def send_data(self, filename, timeout): 
    destination='foo' 
    logging.log(logging.DEBUG, 'sending_data') 
    tasks = [] 
    with open(filename) as log_file: 
     for line in log_file: 
      try: 
       json_event = json.loads(line) 
      except ValueError as e: 
       print("Error parsing json event") 
      yield from asyncio.sleep(timeout) 
      tasks.append(asyncio.async(
       self.handle_line(json.dumps(json_event), destination)) 
    yield from asyncio.wait(tasks) 


asyncio.get_event_loop().run_until_complete(send_data('foo.txt', 1))