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))