zoukankan      html  css  js  c++  java

  • asyncio 学习

    来自:https://www.syncd.cn/article/asyncio_article_02

     

     

    一、asyncio之—-入门初探

    通过上一篇关于asyncio的整体介绍，看过之后基本对asyncio就有一个基本认识，如果是感兴趣的小伙伴相信也会尝试写一些小代码尝试用了，那么这篇文章会通过一个简单的爬虫程序，从简单到复杂，一点一点的改进程序以达到我们想要的效果.

    https://github.com/HackerNews/API 这里是关于HN的API的使用说明，这次写的爬虫就是调用这里的api接口，用到的模块是aiohttp 发起的请求，切记这里是不能用requests模块的。关于aiohttp的文档：https://aiohttp.readthedocs.io/en/stable/

    下面我们看具体的代码实现，这个代码主要就是爬取其中一个连接下的所有评论，如果不传递id的情况，默认就是爬取id为8863的评论

    1. import asyncio
    2. import argparse
    3. import logging
    4. from urllib.parse import urlparse, parse_qs
    5. from datetime import datetime
    7. import aiohttp
    8. import async_timeout
    11. LOGGER_FORMAT = '%(asctime)s %(message)s'
    12. URL_TEMPLATE = "https://hacker-news.firebaseio.com/v0/item/{}.json"
    13. FETCH_TIMEOUT = 10
    15. parser = argparse.ArgumentParser(
    16. description='获取所有请求url的所有评论')
    17. parser.add_argument('--id', type=int, default=8863,
    18. help='请求的id, 默认id 是8863')
    19. parser.add_argument('--url', type=str, help='HN的url地址')
    20. parser.add_argument('--verbose', action='store_true', help='详细的输出')
    23. logging.basicConfig(format=LOGGER_FORMAT, datefmt='[%H:%M:%S]')
    24. log = logging.getLogger()
    25. log.setLevel(logging.INFO)
    27. fetch_counter = 0
    30. async def fetch(session, url):
    31. """
    32. 通过aiohttp访问url并返回json格式数据
    33. """
    34. global fetch_counter
    35. with async_timeout.timeout(FETCH_TIMEOUT):
    36. fetch_counter += 1
    37. # 因为接口需要翻墙才能访问，所以这里我用的是我本地的代理
    38. async with session.get(url,proxy='http://127.0.0.1:1081') as response:
    39. return await response.json()
    42. async def post_number_of_comments(loop, session, post_id):
    43. """
    44. 递归获取当前请求url的所有评论
    45. """
    46. url = URL_TEMPLATE.format(post_id)
    47. now = datetime.now()
    48. response = await fetch(session, url)
    49. log.debug('{:^6} > Fetching of {} took {} seconds'.format(
    50. post_id, url, (datetime.now() - now).total_seconds()))
    52. if 'kids' not in response: #表示没有评论
    53. return 0
    55. # 获取当前请求的url的评论的数量
    56. number_of_comments = len(response['kids'])
    58. log.debug('{:^6} > Fetching {} child posts'.format(
    59. post_id, number_of_comments))
    60. tasks = [post_number_of_comments(
    61. loop, session, kid_id) for kid_id in response['kids']]
    63. # 获取所有协程的执行的结果
    64. results = await asyncio.gather(*tasks)
    66. number_of_comments += sum(results)
    67. log.debug('{:^6} > {} comments'.format(post_id, number_of_comments))
    69. return number_of_comments
    72. def id_from_HN_url(url):
    73. """
    74. 获取运行时传递的参数中的id
    75. """
    76. parse_result = urlparse(url)
    77. try:
    78. return parse_qs(parse_result.query)['id'][0]
    79. except (KeyError, IndexError):
    80. return None
    83. async def main(loop, post_id):
    84. now = datetime.now()
    85. async with aiohttp.ClientSession(loop=loop) as session:
    86. now = datetime.now()
    87. comments = await post_number_of_comments(loop, session, post_id)
    88. log.info(
    89. '> Calculating comments took {:.2f} seconds and {} fetches'.format(
    90. (datetime.now() - now).total_seconds(), fetch_counter))
    92. return comments
    95. if __name__ == '__main__':
    96. args = parser.parse_args()
    97. if args.verbose:
    98. log.setLevel(logging.DEBUG)
    100. post_id = id_from_HN_url(args.url) if args.url else args.id
    102. loop = asyncio.get_event_loop()
    103. comments = loop.run_until_complete(main(loop, post_id))
    104. log.info("-- Post {} has {} comments".format(post_id, comments))
    106. loop.close()

    再次提醒该url请求的时候是需要翻墙才能访问到，所以我这里加了本地的代理，以便能够爬取到内容，正常的请求结果如下：

    1. [23:24:37] > Calculating comments took 2.98 seconds and 73 fetches
    2. [23:24:37] -- Post 8863 has 72 comments

    如果没有翻墙就是如下错误了：

    1. Traceback (most recent call last):
    2. File "/Users/zhaofan/vs_python/python_asyncio/ex1.py", line 41, in fetch
    3. async with session.get(url) as response:
    4. File "/usr/local/lib/python3.7/site-packages/aiohttp/client.py", line 1005, in __aenter__
    5. self._resp = await self._coro
    6. File "/usr/local/lib/python3.7/site-packages/aiohttp/client.py", line 476, in _request
    7. timeout=real_timeout
    8. File "/usr/local/lib/python3.7/site-packages/aiohttp/connector.py", line 522, in connect
    9. proto = await self._create_connection(req, traces, timeout)
    10. File "/usr/local/lib/python3.7/site-packages/aiohttp/connector.py", line 854, in _create_connection
    11. req, traces, timeout)
    12. File "/usr/local/lib/python3.7/site-packages/aiohttp/connector.py", line 974, in _create_direct_connection
    13. req=req, client_error=client_error)
    14. File "/usr/local/lib/python3.7/site-packages/aiohttp/connector.py", line 924, in _wrap_create_connection
    15. await self._loop.create_connection(*args, **kwargs))
    16. File "/usr/local/Cellar/python/3.7.2_2/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 946, in create_connection
    17. await self.sock_connect(sock, address)
    18. File "/usr/local/Cellar/python/3.7.2_2/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/selector_events.py", line 464, in sock_connect
    19. return await fut
    20. concurrent.futures._base.CancelledError
    22. During handling of the above exception, another exception occurred:
    24. Traceback (most recent call last):
    25. File "/Users/zhaofan/vs_python/python_asyncio/ex1.py", line 115, in <module>
    26. comments = loop.run_until_complete(main(loop, post_id))
    27. File "/usr/local/Cellar/python/3.7.2_2/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    28. return future.result()
    29. File "/Users/zhaofan/vs_python/python_asyncio/ex1.py", line 99, in main
    30. comments = await post_number_of_comments(loop, session, post_id)
    31. File "/Users/zhaofan/vs_python/python_asyncio/ex1.py", line 51, in post_number_of_comments
    32. response = await fetch(session, url)
    33. File "/Users/zhaofan/vs_python/python_asyncio/ex1.py", line 42, in fetch
    34. return await response.json()
    35. File "/usr/local/lib/python3.7/site-packages/async_timeout/__init__.py", line 45, in __exit__
    36. self._do_exit(exc_type)
    37. File "/usr/local/lib/python3.7/site-packages/async_timeout/__init__.py", line 92, in _do_exit
    38. raise asyncio.TimeoutError
    39. concurrent.futures._base.TimeoutError

    还有就是上面的代码中我们使用了results = await asyncio.gather(*tasks)
    等待所有的协程执行完成并返回结果，关于gather的官网文档地址：https://docs.python.org/3/library/asyncio-task.html#asyncio.gather

    并且在上面的使用中我们也用到了递归，你可能感觉还挺简单的，代码看着和我们平时的写的阻塞式的代码好像区别也不是特别大，保持这种愉悦感，接着看

    二、asyncio之—-更进一步

    那么我们现在想要的是当我们的爬虫程序爬取评论的时候，我们想要当评论超过一定阈值的贴帖子发邮件通知告诉我们，其实这个功能是非常有必要的，就拿我的个人博客站来说，如果你想要经常看我的文章，又不想经常来我的站看，只想看大家都关注的那些文章，或者评论比较多的文章，所以我们接着将我们的代码进行更改：

    1. import asyncio
    2. import argparse
    3. import logging
    4. import random
    5. from urllib.parse import urlparse, parse_qs
    6. from datetime import datetime
    8. import aiohttp
    9. import async_timeout
    12. LOGGER_FORMAT = '%(asctime)s %(message)s'
    13. URL_TEMPLATE = "https://hacker-news.firebaseio.com/v0/item/{}.json"
    14. FETCH_TIMEOUT = 10
    15. # 我们设置的评论的阈值
    16. MIN_COMMENTS = 2
    18. parser = argparse.ArgumentParser(
    19. description='获取所有请求url的所有评论')
    20. parser.add_argument('--id', type=int, default=8863,
    21. help='请求的id, 默认id 是8863')
    22. parser.add_argument('--url', type=str, help='HN的url地址')
    23. parser.add_argument('--verbose', action='store_true', help='详细的输出')
    26. logging.basicConfig(format=LOGGER_FORMAT, datefmt='[%H:%M:%S]')
    27. log = logging.getLogger()
    28. log.setLevel(logging.INFO)
    30. fetch_counter = 0
    33. async def fetch(session, url):
    34. """
    35. 通过aiohttp访问url并返回json格式数据
    36. """
    37. global fetch_counter
    38. with async_timeout.timeout(FETCH_TIMEOUT):
    39. fetch_counter += 1
    40. # 因为接口需要翻墙才能访问，所以这里我用的是我本地的代理
    41. async with session.get(url,proxy='http://127.0.0.1:1081') as response:
    42. return await response.json()
    45. async def post_number_of_comments(loop, session, post_id):
    46. """
    47. 递归获取当前请求url的所有评论
    48. """
    49. url = URL_TEMPLATE.format(post_id)
    50. now = datetime.now()
    51. response = await fetch(session, url)
    52. log.debug('{:^6} > Fetching of {} took {} seconds'.format(
    53. post_id, url, (datetime.now() - now).total_seconds()))
    55. if 'kids' not in response: #表示没有评论
    56. return 0
    58. # 获取当前请求的url的评论的数量
    59. number_of_comments = len(response['kids'])
    61. log.debug('{:^6} > Fetching {} child posts'.format(
    62. post_id, number_of_comments))
    63. tasks = [post_number_of_comments(
    64. loop, session, kid_id) for kid_id in response['kids']]
    66. # 获取所有任务的结果
    67. results = await asyncio.gather(*tasks)
    69. number_of_comments += sum(results)
    70. log.debug('{:^6} > {} comments'.format(post_id, number_of_comments))
    71. if number_of_comments> MIN_COMMENTS:
    72. await email_post(response)
    73. return number_of_comments
    75. async def email_post(post):
    76. """
    77. 模拟发邮件的动作，并没有真的发邮件
    78. """
    79. await asyncio.sleep(random.random()*3)
    80. log.info("email send success")
    83. def id_from_HN_url(url):
    84. """
    85. 获取运行时传递的参数中的id
    86. """
    87. parse_result = urlparse(url)
    88. try:
    89. return parse_qs(parse_result.query)['id'][0]
    90. except (KeyError, IndexError):
    91. return None
    94. async def main(loop, post_id):
    95. now = datetime.now()
    96. async with aiohttp.ClientSession(loop=loop) as session:
    97. now = datetime.now()
    98. comments = await post_number_of_comments(loop, session, post_id)
    99. log.info(
    100. '> Calculating comments took {:.2f} seconds and {} fetches'.format(
    101. (datetime.now() - now).total_seconds(), fetch_counter))
    103. return comments
    106. if __name__ == '__main__':
    107. args = parser.parse_args()
    108. if args.verbose:
    109. log.setLevel(logging.DEBUG)
    111. post_id = id_from_HN_url(args.url) if args.url else args.id
    113. loop = asyncio.get_event_loop()
    114. comments = loop.run_until_complete(main(loop, post_id))
    115. log.info("-- Post {} has {} comments".format(post_id, comments))
    117. loop.close()

    运行结果如下：

    1. [23:24:17] email send success
    2. [23:24:18] email send success
    3. [23:24:18] email send success
    4. [23:24:19] email send success
    5. [23:24:19] email send success
    6. [23:24:20] email send success
    7. [23:24:21] email send success
    8. [23:24:21] email send success
    9. [23:24:24] email send success
    10. [23:24:24] > Calculating comments took 10.09 seconds and 73 fetches
    11. [23:24:24] -- Post 8863 has 72 comments

    你会发现这次花费的时间比我们之前多了，因为我们在发送邮件的地方是 await email_post(response) 那么我们的的程序再这里就会等到知道这个任务完成，其实对我们来说我们更关注的是我们的主要任务，获取所有的评论结果，而发送邮件通知我们的次级任务，那么我们需要怎么改进，让我们的主要的任务继续执行，不用去等待子任务的执行呢？在asyncio的api文档中有ensure_future ，这个需要注意：在python3.7之前用的是这个方法，但3.7之后更推荐用create_task的方法 具体地址为：https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task
    这里明确说明了：

    asyncio.create_task(coro)
    Wrap the coro coroutine into a Task and schedule its execution. Return the Task object.
    The task is executed in the loop returned by get_running_loop(), RuntimeError is raised if there is no running loop in current thread.
    This function has been added in Python 3.7. Prior to Python 3.7, the low-level asyncio.ensure_future() function can be used instead:

    通过这个方法我们可以将我们的任务安排一个协程运行，将其包装在Task对象中并返回它，既然这样我们就将代码继续更改：
    将await email_post(response) 这样代码替换为：asyncio.ensure_future(email_post(response))

    但是当我们运行后发现不幸的事情发生了：

    1. [23:40:06] email send success
    2. [23:40:06] > Calculating comments took 3.30 seconds and 73 fetches
    3. [23:40:06] -- Post 8863 has 72 comments
    4. [23:40:06] Task was destroyed but it is pending!
    5. task: <Task pending coro=<email_post() done, defined at ex1.py:76> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x1087dde58>()]>>
    6. [23:40:06] Task was destroyed but it is pending!
    7. task: <Task pending coro=<email_post() done, defined at ex1.py:76> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x108a9e4f8>()]>>
    8. [23:40:06] Task was destroyed but it is pending!
    9. task: <Task pending coro=<email_post() done, defined at ex1.py:76> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x108a9e9a8>()]>>
    10. [23:40:06] Task was destroyed but it is pending!
    11. task: <Task pending coro=<email_post() done, defined at ex1.py:76> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x108a9e918>()]>>
    12. [23:40:06] Task was destroyed but it is pending!
    13. task: <Task pending coro=<email_post() done, defined at ex1.py:76> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x108a9ee88>()]>>
    14. [23:40:06] Task was destroyed but it is pending!
    15. task: <Task pending coro=<email_post() done, defined at ex1.py:76> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x108a9ef48>()]>>
    16. [23:40:06] Task was destroyed but it is pending!
    17. task: <Task pending coro=<email_post() done, defined at ex1.py:76> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x108a9efd8>()]>>
    18. [23:40:06] Task was destroyed but it is pending!
    19. task: <Task pending coro=<email_post() done, defined at ex1.py:76> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x1087dde28>()]>>

    看到这个错误不要慌，这个也是很多初学asyncio的或者刚开始用的时候都会碰到的问题，并且这个问题我们在上一篇asyncio的文章也说明了原因，在这里其实就是post_number_of_comments协程返回后立即强行关闭循环，让我们的log_post任务没有时间完成,怎么解决呢？ 我们继续改代码：

    1. import asyncio
    2. import argparse
    3. import logging
    4. import random
    5. from urllib.parse import urlparse, parse_qs
    6. from datetime import datetime
    8. import aiohttp
    9. import async_timeout
    12. LOGGER_FORMAT = '%(asctime)s %(message)s'
    13. URL_TEMPLATE = "https://hacker-news.firebaseio.com/v0/item/{}.json"
    14. FETCH_TIMEOUT = 10
    15. # 我们设置的评论的阈值
    16. MIN_COMMENTS = 2
    18. parser = argparse.ArgumentParser(
    19. description='获取所有请求url的所有评论')
    20. parser.add_argument('--id', type=int, default=8863,
    21. help='请求的id, 默认id 是8863')
    22. parser.add_argument('--url', type=str, help='HN的url地址')
    23. parser.add_argument('--verbose', action='store_true', help='详细的输出')
    26. logging.basicConfig(format=LOGGER_FORMAT, datefmt='[%H:%M:%S]')
    27. log = logging.getLogger()
    28. log.setLevel(logging.INFO)
    30. fetch_counter = 0
    33. async def fetch(session, url):
    34. """
    35. 通过aiohttp访问url并返回json格式数据
    36. """
    37. global fetch_counter
    38. with async_timeout.timeout(FETCH_TIMEOUT):
    39. fetch_counter += 1
    40. # 因为接口需要翻墙才能访问，所以这里我用的是我本地的代理
    41. async with session.get(url,proxy='http://127.0.0.1:1081') as response:
    42. return await response.json()
    45. async def post_number_of_comments(loop, session, post_id):
    46. """
    47. 递归获取当前请求url的所有评论
    48. """
    49. url = URL_TEMPLATE.format(post_id)
    50. now = datetime.now()
    51. response = await fetch(session, url)
    52. log.debug('{:^6} > Fetching of {} took {} seconds'.format(
    53. post_id, url, (datetime.now() - now).total_seconds()))
    55. if 'kids' not in response: #表示没有评论
    56. return 0
    58. # 获取当前请求的url的评论的数量
    59. number_of_comments = len(response['kids'])
    61. log.debug('{:^6} > Fetching {} child posts'.format(
    62. post_id, number_of_comments))
    63. tasks = [post_number_of_comments(
    64. loop, session, kid_id) for kid_id in response['kids']]
    66. # 获取所有任务的结果
    67. results = await asyncio.gather(*tasks)
    69. number_of_comments += sum(results)
    70. log.debug('{:^6} > {} comments'.format(post_id, number_of_comments))
    71. if number_of_comments> MIN_COMMENTS:
    72. # await email_post(response)
    73. asyncio.ensure_future(email_post(response))
    74. return number_of_comments
    76. async def email_post(post):
    77. """
    78. 模拟发邮件的动作，并没有真的发邮件
    79. """
    80. await asyncio.sleep(random.random()*3)
    81. log.info("email send success")
    84. def id_from_HN_url(url):
    85. """
    86. 获取运行时传递的参数中的id
    87. """
    88. parse_result = urlparse(url)
    89. try:
    90. return parse_qs(parse_result.query)['id'][0]
    91. except (KeyError, IndexError):
    92. return None
    95. async def main(loop, post_id):
    96. now = datetime.now()
    97. async with aiohttp.ClientSession(loop=loop) as session:
    98. now = datetime.now()
    99. comments = await post_number_of_comments(loop, session, post_id)
    100. log.info(
    101. '> Calculating comments took {:.2f} seconds and {} fetches'.format(
    102. (datetime.now() - now).total_seconds(), fetch_counter))
    104. return comments
    107. if __name__ == '__main__':
    108. args = parser.parse_args()
    109. if args.verbose:
    110. log.setLevel(logging.DEBUG)
    112. post_id = id_from_HN_url(args.url) if args.url else args.id
    114. loop = asyncio.get_event_loop()
    115. comments = loop.run_until_complete(main(loop, post_id))
    116. log.info("-- Post {} has {} comments".format(post_id, comments))
    118. pending_tasks = [
    119. task for task in asyncio.Task.all_tasks() if not task.done()
    120. ]
    121. loop.run_until_complete(asyncio.gather(*pending_tasks))
    123. loop.close()

    运行之后结果如下：

    1. [23:47:24] email send success
    2. [23:47:25] email send success
    3. [23:47:25] > Calculating comments took 3.29 seconds and 73 fetches
    4. [23:47:25] -- Post 8863 has 72 comments
    5. [23:47:25] email send success
    6. [23:47:25] email send success
    7. [23:47:25] email send success
    8. [23:47:26] email send success
    9. [23:47:26] email send success
    10. [23:47:27] email send success
    11. [23:47:27] email send success

    一切似乎又恢复了正常，这里我们用到了asyncio的一个方法
    asyncio.Task.all_tasks()

    这个其实还是非常有用的可以获取当前我们的loop的所有的任务的情况，我们这里是通过task.done() 来判断任务是否完成了，从而把没有让没有完成的任务都能够继续完成，但是我们这样做有一个不好的地方就是asyncio.Task.all_tasks() 将所有的任务都拿到手了，可是有些并不是我们关注的，我们就只想要控制我们自己关注的，那么我们就可以将发邮件这个次级任务专门放到一起，这样方面我们后面处理，代码更改为：

    1. import asyncio
    2. import argparse
    3. import logging
    4. import random
    5. from urllib.parse import urlparse, parse_qs
    6. from datetime import datetime
    8. import aiohttp
    9. import async_timeout
    12. LOGGER_FORMAT = '%(asctime)s %(message)s'
    13. URL_TEMPLATE = "https://hacker-news.firebaseio.com/v0/item/{}.json"
    14. FETCH_TIMEOUT = 10
    15. # 我们设置的评论的阈值
    16. MIN_COMMENTS = 2
    18. parser = argparse.ArgumentParser(
    19. description='获取所有请求url的所有评论')
    20. parser.add_argument('--id', type=int, default=8863,
    21. help='请求的id, 默认id 是8863')
    22. parser.add_argument('--url', type=str, help='HN的url地址')
    23. parser.add_argument('--verbose', action='store_true', help='详细的输出')
    26. logging.basicConfig(format=LOGGER_FORMAT, datefmt='[%H:%M:%S]')
    27. log = logging.getLogger()
    28. log.setLevel(logging.INFO)
    30. fetch_counter = 0
    33. async def fetch(session, url):
    34. """
    35. 通过aiohttp访问url并返回json格式数据
    36. """
    37. global fetch_counter
    38. with async_timeout.timeout(FETCH_TIMEOUT):
    39. fetch_counter += 1
    40. # 因为接口需要翻墙才能访问，所以这里我用的是我本地的代理
    41. async with session.get(url,proxy='http://127.0.0.1:1081') as response:
    42. return await response.json()
    45. async def post_number_of_comments(loop, session, post_id):
    46. """
    47. 递归获取当前请求url的所有评论
    48. """
    49. url = URL_TEMPLATE.format(post_id)
    50. now = datetime.now()
    51. response = await fetch(session, url)
    52. log.debug('{:^6} > Fetching of {} took {} seconds'.format(
    53. post_id, url, (datetime.now() - now).total_seconds()))
    55. if 'kids' not in response: #表示没有评论
    56. return 0
    58. # 获取当前请求的url的评论的数量
    59. number_of_comments = len(response['kids'])
    61. log.debug('{:^6} > Fetching {} child posts'.format(
    62. post_id, number_of_comments))
    63. tasks = [post_number_of_comments(
    64. loop, session, kid_id) for kid_id in response['kids']]
    66. # 获取所有任务的结果
    67. results = await asyncio.gather(*tasks)
    69. number_of_comments += sum(results)
    70. log.debug('{:^6} > {} comments'.format(post_id, number_of_comments))
    71. if number_of_comments> MIN_COMMENTS:
    72. # await email_post(response)
    73. task_registry.append(asyncio.ensure_future(email_post(response)))
    74. return number_of_comments
    76. async def email_post(post):
    77. """
    78. 模拟发邮件的动作，并没有真的发邮件
    79. """
    80. await asyncio.sleep(random.random()*3)
    81. log.info("email send success")
    84. def id_from_HN_url(url):
    85. """
    86. 获取运行时传递的参数中的id
    87. """
    88. parse_result = urlparse(url)
    89. try:
    90. return parse_qs(parse_result.query)['id'][0]
    91. except (KeyError, IndexError):
    92. return None
    95. async def main(loop, post_id):
    96. now = datetime.now()
    97. async with aiohttp.ClientSession(loop=loop) as session:
    98. now = datetime.now()
    99. comments = await post_number_of_comments(loop, session, post_id)
    100. log.info(
    101. '> Calculating comments took {:.2f} seconds and {} fetches'.format(
    102. (datetime.now() - now).total_seconds(), fetch_counter))
    104. return comments
    107. if __name__ == '__main__':
    108. args = parser.parse_args()
    109. if args.verbose:
    110. log.setLevel(logging.DEBUG)
    112. post_id = id_from_HN_url(args.url) if args.url else args.id
    113. task_registry = [] # 用于存放我们发送邮件的次级任务
    114. loop = asyncio.get_event_loop()
    115. comments = loop.run_until_complete(main(loop, post_id))
    116. log.info("-- Post {} has {} comments".format(post_id, comments))
    118. pending_tasks = [
    119. task for task in task_registry if not task.done()
    120. ]
    121. loop.run_until_complete(asyncio.gather(*pending_tasks))
    123. loop.close()

    执行结果如下：

    1. [23:54:10] > Calculating comments took 8.33 seconds and 73 fetches
    2. [23:54:10] -- Post 8863 has 72 comments
    3. [23:54:11] email send success
    4. [23:54:11] email send success
    5. [23:54:11] email send success
    6. [23:54:12] email send success
    7. [23:54:12] email send success
    8. [23:54:12] email send success
    9. [23:54:12] email send success
    10. [23:54:13] email send success
    11. [23:54:13] email send success

    看到这里，你是不是发现其实python的asyncio也没有那么难，貌似还挺好用的，那么我们接着最后一部分

    asyncio之—-华山论剑

    通过上面的代码的不断改进, 我们也渐渐更加熟悉asyncio 的用法,但是相对来说还是太简单，因为到目前为止，我们都在爬取一个url 下的所有评论，那么如果我想要获取多个url下的评论信息需要怎么改进代码。在HN 的API文档中有一个获取top 500的接口, 那么我们只获取前500中的前几个的所有评论，当然这个top 500 的内容每天肯能都会更新，甚至可能一天之内都会更新，所以我们的任务需要可以获取一次之后过一会再次获取一次数据，这样我们就能总是获取最新的数据，我们将代码继续改进:

    1. import asyncio
    2. import argparse
    3. import logging
    4. from datetime import datetime
    6. import aiohttp
    7. import async_timeout
    10. LOGGER_FORMAT = '%(asctime)s %(message)s'
    11. URL_TEMPLATE = "https://hacker-news.firebaseio.com/v0/item/{}.json"
    12. TOP_STORIES_URL = "https://hacker-news.firebaseio.com/v0/topstories.json"
    13. FETCH_TIMEOUT = 10
    15. parser = argparse.ArgumentParser(
    16. description='获取Hacker News 文章的评论数')
    17. parser.add_argument(
    18. '--period', type=int, default=5, help='每个任务的间隔时间')
    19. parser.add_argument(
    20. '--limit', type=int, default=5,
    21. help='获取top 500的前n 数量内容默认是前500的前5个')
    22. parser.add_argument('--verbose', action='store_true', help='更加详细的输出')
    25. logging.basicConfig(format=LOGGER_FORMAT, datefmt='[%H:%M:%S]')
    26. log = logging.getLogger()
    27. log.setLevel(logging.INFO)
    29. fetch_counter = 0
    32. async def fetch(session, url):
    33. """
    34. 请求url地址返回json格式数据
    35. """
    36. global fetch_counter
    37. with async_timeout.timeout(FETCH_TIMEOUT):
    38. fetch_counter += 1
    39. async with session.get(url, proxy="http://127.0.0.1:1080") as response:
    40. return await response.json()
    43. async def post_number_of_comments(loop, session, post_id):
    44. """
    45. 获取当前文章的数据，并递归获取所有评论
    46. """
    47. url = URL_TEMPLATE.format(post_id)
    48. now = datetime.now()
    49. response = await fetch(session, url)
    50. log.debug('{:^6} > Fetching of {} took {} seconds'.format(
    51. post_id, url, (datetime.now() - now).total_seconds()))
    53. if 'kids' not in response: # 没有评论
    54. return 0
    56. # 获取当前文章的评论数量
    57. number_of_comments = len(response['kids'])
    59. log.debug('{:^6} > Fetching {} child posts'.format(
    60. post_id, number_of_comments))
    62. tasks = [post_number_of_comments(
    63. loop, session, kid_id) for kid_id in response['kids']]
    65. # 这里递归请求获取每条评论的评论
    66. results = await asyncio.gather(*tasks)
    68. # 获取当前文章的总评论数
    69. number_of_comments += sum(results)
    70. log.debug('{:^6} > {} comments'.format(post_id, number_of_comments))
    71. return number_of_comments
    74. async def get_comments_of_top_stories(loop, session, limit, iteration):
    75. """
    76. 获取top 500de 前5个
    77. """
    78. response = await fetch(session, TOP_STORIES_URL)
    79. tasks = [post_number_of_comments(
    80. loop, session, post_id) for post_id in response[:limit]]
    81. results = await asyncio.gather(*tasks)
    82. for post_id, num_comments in zip(response[:limit], results):
    83. log.info("Post {} has {} comments ({})".format(
    84. post_id, num_comments, iteration))
    87. async def poll_top_stories_for_comments(loop, session, period, limit):
    88. """
    89. 定时去请求获取前top 500 url
    90. """
    91. global fetch_counter
    92. iteration = 1
    93. while True:
    94. now = datetime.now()
    95. log.info("Calculating comments for top {} stories. ({})".format(
    96. limit, iteration))
    98. await get_comments_of_top_stories(loop, session, limit, iteration)
    100. log.info(
    101. '> Calculating comments took {:.2f} seconds and {} fetches'.format(
    102. (datetime.now() - now).total_seconds(), fetch_counter))
    103. log.info("Waiting for {} seconds...".format(period))
    104. iteration += 1
    105. fetch_counter = 0
    106. # 每个任务的间隔
    107. await asyncio.sleep(period)
    110. async def main(loop, period, limit):
    111. async with aiohttp.ClientSession(loop=loop) as session:
    112. comments = await poll_top_stories_for_comments(loop, session, period, limit)
    114. return comments
    117. if __name__ == '__main__':
    118. args = parser.parse_args()
    119. if args.verbose:
    120. log.setLevel(logging.DEBUG)
    122. loop = asyncio.get_event_loop()
    123. loop.run_until_complete(main(loop, args.period, args.limit))
    125. loop.close()

    查看运行结果如下：

    1. [16:24:28] Calculating comments for top 5 stories. (1)
    2. [16:24:41] Post 19334909 has 156 comments (1)
    3. [16:24:41] Post 19333600 has 147 comments (1)
    4. [16:24:41] Post 19335363 has 9 comments (1)
    5. [16:24:41] Post 19330812 has 341 comments (1)
    6. [16:24:41] Post 19333479 has 81 comments (1)
    7. [16:24:41] > Calculating comments took 12.17 seconds and 740 fetches
    8. [16:24:41] Waiting for 5 seconds...
    9. [16:24:46] Calculating comments for top 5 stories. (2)
    10. [16:24:50] Post 19334909 has 156 comments (2)
    11. [16:24:50] Post 19333600 has 147 comments (2)
    12. [16:24:50] Post 19335363 has 9 comments (2)
    13. [16:24:50] Post 19330812 has 341 comments (2)
    14. [16:24:50] Post 19333479 has 81 comments (2)
    15. [16:24:50] > Calculating comments took 4.75 seconds and 740 fetches
    16. [16:24:50] Waiting for 5 seconds...
    17. Traceback (most recent call last):

    运行结果我们看出来其实我们的每个任务并不是间隔5s,因为我的任务在 await get_comments_of_top_stories(loop, session, limit, iteration)
    我们必须等到这个地方完成之后才会进入下次循环，但是其实有时候我们并不想等待，而是直接想要继续往下走，那么我们还是通过老办法通过ensure_future 实现,我们将那一行代码更改为：

    asyncio.ensure_future(get_comments_of_top_stories(loop, session, limit, iteration))

    再次运行结果之后：

    1. [16:44:07] Calculating comments for top 5 stories. (1)
    2. [16:44:07] > Calculating comments took 0.00 seconds and 0 fetches
    3. [16:44:07] Waiting for 5 seconds...
    4. [16:44:12] Calculating comments for top 5 stories. (2)
    5. [16:44:12] > Calculating comments took 0.00 seconds and 49 fetches
    6. [16:44:12] Waiting for 5 seconds...
    7. [16:44:17] Calculating comments for top 5 stories. (3)
    8. [16:44:17] > Calculating comments took 0.00 seconds and 1044 fetches
    9. [16:44:17] Waiting for 5 seconds...
    10. [16:44:21] Post 19334909 has 159 comments (1)
    11. [16:44:21] Post 19333600 has 150 comments (1)
    12. [16:44:21] Post 19335363 has 13 comments (1)
    13. [16:44:21] Post 19330812 has 342 comments (1)
    14. [16:44:21] Post 19333479 has 81 comments (1)
    15. [16:44:22] Post 19334909 has 159 comments (3)
    16. [16:44:22] Post 19333600 has 150 comments (3)
    17. [16:44:22] Post 19335363 has 13 comments (3)
    18. [16:44:22] Post 19330812 has 342 comments (3)
    19. [16:44:22] Post 19333479 has 81 comments (3)
    20. [16:44:22] Calculating comments for top 5 stories. (4)
    21. [16:44:22] > Calculating comments took 0.00 seconds and 1158 fetches
    22. [16:44:22] Waiting for 5 seconds...
    23. [16:44:23] Post 19334909 has 159 comments (2)
    24. [16:44:23] Post 19333600 has 150 comments (2)
    25. [16:44:23] Post 19335363 has 13 comments (2)
    26. [16:44:23] Post 19330812 has 342 comments (2)
    27. [16:44:23] Post 19333479 has 81 comments (2)
    28. [16:44:26] Post 19334909 has 159 comments (4)
    29. [16:44:26] Post 19333600 has 150 comments (4)
    30. [16:44:26] Post 19335363 has 13 comments (4)
    31. [16:44:26] Post 19330812 has 343 comments (4)
    32. [16:44:26] Post 19333479 has 81 comments (4)
    33. [16:44:27] Calculating comments for top 5 stories. (5)
    34. [16:44:27] > Calculating comments took 0.00 seconds and 754 fetches
    35. [16:44:27] Waiting for 5 seconds...

    这样我们每次任务的间隔倒是是5s了但是又一个问题出现了，花费0s并且0个fetch到，并且续的fetch数量也都不对 ，其实造成这个的原因都是因为不再等待get_comments_of_top_stories(loop, session, limit, iteration)造成的

    这个时候你是不是又想到了你的老朋友 callback 呢 哈哈哈！ 改进代码如下：

    1. import asyncio
    2. import argparse
    3. import logging
    4. from datetime import datetime
    6. import aiohttp
    7. import async_timeout
    10. LOGGER_FORMAT = '%(asctime)s %(message)s'
    11. URL_TEMPLATE = "https://hacker-news.firebaseio.com/v0/item/{}.json"
    12. TOP_STORIES_URL = "https://hacker-news.firebaseio.com/v0/topstories.json"
    13. FETCH_TIMEOUT = 10
    15. parser = argparse.ArgumentParser(
    16. description='获取Hacker News 文章的评论数')
    17. parser.add_argument(
    18. '--period', type=int, default=5, help='每个任务的间隔时间')
    19. parser.add_argument(
    20. '--limit', type=int, default=5,
    21. help='获取top 500的前n 数量内容默认是前500的前5个')
    22. parser.add_argument('--verbose', action='store_true', help='更加详细的输出')
    25. logging.basicConfig(format=LOGGER_FORMAT, datefmt='[%H:%M:%S]')
    26. log = logging.getLogger()
    27. log.setLevel(logging.INFO)
    30. class URLFetcher():
    32. def __init__(self):
    33. self.fetch_counter = 0
    35. async def fetch(self, session, url):
    36. with async_timeout.timeout(FETCH_TIMEOUT):
    37. self.fetch_counter += 1
    38. async with session.get(url, proxy="http://127.0.0.1:1080") as response:
    39. return await response.json()
    42. async def post_number_of_comments(loop, session, fetcher, post_id):
    43. """
    44. 获取当前文章的数据，并递归获取所有评论
    45. """
    46. url = URL_TEMPLATE.format(post_id)
    47. response = await fetcher.fetch(session, url)
    49. # 没有评论
    50. if response is None or 'kids' not in response:
    51. return 0
    53. number_of_comments = len(response['kids'])
    55. # # 获取当前文章的评论数量
    56. tasks = [post_number_of_comments(
    57. loop, session, fetcher, kid_id) for kid_id in response['kids']]
    59. # 这里递归请求获取每条评论的评论
    60. results = await asyncio.gather(*tasks)
    62. # 获取当前文章的总评论数
    63. number_of_comments += sum(results)
    64. log.debug('{:^6} > {} comments'.format(post_id, number_of_comments))
    66. return number_of_comments
    69. async def get_comments_of_top_stories(loop, session, limit, iteration):
    70. """
    71. 获取top 500de 前5个
    72. """
    73. fetcher = URLFetcher()
    74. response = await fetcher.fetch(session, TOP_STORIES_URL)
    76. tasks = [post_number_of_comments(
    77. loop, session, fetcher, post_id) for post_id in response[:limit]]
    78. results = await asyncio.gather(*tasks)
    79. for post_id, num_comments in zip(response[:limit], results):
    80. log.info("Post {} has {} comments ({})".format(
    81. post_id, num_comments, iteration))
    82. return fetcher.fetch_counter
    85. async def poll_top_stories_for_comments(loop, session, period, limit):
    86. """
    87. 定时去请求获取前top 500 url
    88. """
    89. iteration = 1
    90. while True:
    91. log.info("Calculating comments for top {} stories. ({})".format(
    92. limit, iteration))
    94. future = asyncio.ensure_future(
    95. get_comments_of_top_stories(loop, session, limit, iteration))
    97. now = datetime.now()
    99. # 这里通过回调的方式获取每次爬取评论的耗时以及爬取的评论的数量
    100. def callback(fut):
    101. fetch_count = fut.result()
    102. log.info(
    103. '> Calculating comments took {:.2f} seconds and {} fetches'.format(
    104. (datetime.now() - now).total_seconds(), fetch_count))
    106. future.add_done_callback(callback)
    108. log.info("Waiting for {} seconds...".format(period))
    109. iteration += 1
    110. await asyncio.sleep(period)
    113. async def main(loop, period, limit):
    114. async with aiohttp.ClientSession(loop=loop) as session:
    115. comments = await poll_top_stories_for_comments(loop, session, period, limit)
    117. return comments
    120. if __name__ == '__main__':
    121. args = parser.parse_args()
    122. if args.verbose:
    123. log.setLevel(logging.DEBUG)
    125. loop = asyncio.get_event_loop()
    126. loop.run_until_complete(main(loop, args.period, args.limit))
    128. loop.close()

    这次当我们再次执行代码运行结果如下：

    1. [17:00:17] Calculating comments for top 5 stories. (1)
    2. [17:00:17] Waiting for 5 seconds...
    3. [17:00:22] Calculating comments for top 5 stories. (2)
    4. [17:00:22] Waiting for 5 seconds...
    5. [17:00:27] Calculating comments for top 5 stories. (3)
    6. [17:00:27] Waiting for 5 seconds...
    7. [17:00:30] Post 19334909 has 163 comments (1)
    8. [17:00:30] Post 19333600 has 152 comments (1)
    9. [17:00:30] Post 19335363 has 14 comments (1)
    10. [17:00:30] Post 19330812 has 346 comments (1)
    11. [17:00:30] Post 19335853 has 1 comments (1)
    12. [17:00:30] > Calculating comments took 2.31 seconds and 682 fetches
    13. [17:00:32] Calculating comments for top 5 stories. (4)
    14. [17:00:32] Waiting for 5 seconds...
    15. [17:00:33] Post 19334909 has 163 comments (2)
    16. [17:00:33] Post 19333600 has 152 comments (2)
    17. [17:00:33] Post 19335363 has 14 comments (2)
    18. [17:00:33] Post 19330812 has 346 comments (2)
    19. [17:00:33] Post 19335853 has 1 comments (2)
    20. [17:00:33] > Calculating comments took 0.80 seconds and 682 fetches
    21. [17:00:34] Post 19334909 has 163 comments (3)
    22. [17:00:34] Post 19333600 has 152 comments (3)
    23. [17:00:34] Post 19335363 has 14 comments (3)
    24. [17:00:34] Post 19330812 has 346 comments (3)
    25. [17:00:34] Post 19335853 has 1 comments (3)
    26. [17:00:34] > Calculating comments took 1.24 seconds and 682 fetches
    27. [17:00:37] Calculating comments for top 5 stories. (5)
    28. [17:00:37] Waiting for 5 seconds...
    29. [17:00:42] Post 19334909 has 163 comments (5)
    30. [17:00:42] Post 19333600 has 152 comments (5)
    31. [17:00:42] Post 19335363 has 15 comments (5)
    32. [17:00:42] Post 19330812 has 346 comments (5)
    33. [17:00:42] Post 19335853 has 1 comments (5)
    34. [17:00:42] > Calculating comments took 4.55 seconds and 683 fetches
    35. [17:00:42] Calculating comments for top 5 stories. (6)
    36. [17:00:42] Waiting for 5 seconds...

    到这里为止，我们的代码基本已经改的可以了，我们的结果也终于达到了一个我们满意的结果。

    小结

    其实对我个人来说，在整理整理之前我自己对asyncio的用法也有很多地方理解的不清楚，也是摸着石头过河，碰到问题解决问题，在整理的过程中，其实对我自己来说很多之前模糊的地方也清晰了很多。
  • 相关阅读:
    pytorch之dataloader深入剖析
    Pytorch多GPU并行处理
    鸡汤：做灯泡的
    数据工程师
    软实力
    七拼八凑的读书感
    深入一点儿
    Session的生命周期
    cloudera项目源代码
    Storm实时计算系统
  • 原文地址：https://www.cnblogs.com/pythonClub/p/10498046.html
Copyright © 2011-2022 走看看