一日一技：使用Asyncio如何限制协程的并发数

在昨天的直播中，有同学问道，如果使用 asyncio + httpx 实现并发请求，怎么限制请求的频率呢?怎么限制最多只能有 x 个请求同时发出呢?我们今天给出两种方案。

提出问题

假设如果我们同时发起12个请求，每个请求的时间不同，那么总共的请求时间大概跟最长耗时的请求差不多。我们先来写一个用于测试的例子：

 
 
 
 
    
  
  
  
  
  1. import asyncio
    2. 
  
  
  
  
  2. import httpx
    3. 
  
  
  
  
  3. import time
    4. 
  
  
  
  
  4. 
  
  
  
  
  5. 
  
  
  
  
  6. async def req(delay):
    7. 
  
  
  
  
  7.     print(f'请求一个延迟为{delay}秒的接口')
    8. 
  
  
  
  
  8.     async with httpx.AsyncClient(timeout=20) as client:
    9. 
  
  
  
  
  9.         resp = await client.get(f'http://127.0.0.1:8000/sleep/{delay}')
    10. 
  
  
  
  
  10.         result = resp.json()
    11. 
  
  
  
  
  11.         print(result)
    12. 
  
  
  
  
  12. 
  
  
  
  
  13. 
  
  
  
  
  14. async def main():
    15. 
  
  
  
  
  15.     start = time.time()
    16. 
  
  
  
  
  16.     delay_list = [3, 6, 1, 8, 2, 4, 5, 2, 7, 3, 9, 8]
    17. 
  
  
  
  
  17.     task_list = []
    18. 
  
  
  
  
  18.     for delay in delay_list:
    19. 
  
  
  
  
  19.         task = asyncio.create_task(req(delay))
    20. 
  
  
  
  
  20.         task_list.append(task)
    21. 
  
  
  
  
  21.     await asyncio.gather(*task_list)
    22. 
  
  
  
  
  22.     end = time.time()
    23. 
  
  
  
  
  23.     print(f'一共耗时：{end - start}')
    24. 
  
  
  
  
  24. 
  
  
  
  
  25. asyncio.run(main())
    26.

这段代码，使用 for 循环创建了12个协程任务，这些任务几乎同时运行，于是，请求完成所有的接口，总共耗时如下图所示：

现在的问题是，由于网站有反爬虫机制，最多只能同时发起3个请求。那么我们怎么确保同一时间最多只有3个协程在请求网络呢?

限制协程任务数

第一个方案跟以前限制多线程的线程数的方案相同。我们创建一个列表，确保列表里面最多只有3个任务，然后持续循环检查，发现有任务完成了，就移除这个完成的任务，并加入一个新的任务，直到待爬的列表为空，这个任务列表也为空。代码如下：

 
 
 
 
    
  
  
  
  
  1. import asyncio
    2. 
  
  
  
  
  2. import httpx
    3. 
  
  
  
  
  3. import time
    4. 
  
  
  
  
  4. 
  
  
  
  
  5. 
  
  
  
  
  6. async def req(delay):
    7. 
  
  
  
  
  7.     print(f'请求一个延迟为{delay}秒的接口')
    8. 
  
  
  
  
  8.     async with httpx.AsyncClient(timeout=20) as client:
    9. 
  
  
  
  
  9.         resp = await client.get(f'http://127.0.0.1:8000/sleep/{delay}')
    10. 
  
  
  
  
  10.         result = resp.json()
    11. 
  
  
  
  
  11.         print(result)
    12. 
  
  
  
  
  12. 
  
  
  
  
  13. 
  
  
  
  
  14. async def main():
    15. 
  
  
  
  
  15.     start = time.time()
    16. 
  
  
  
  
  16.     delay_list = [3, 6, 1, 8, 2, 4, 5, 2, 7, 3, 9, 8]
    17. 
  
  
  
  
  17.     task_list = []
    18. 
  
  
  
  
  18.     while True:
    19. 
  
  
  
  
  19.         if not delay_list and not task_list:
    20. 
  
  
  
  
  20.             break
    21. 
  
  
  
  
  21.         while len(task_list) < 3:
    22. 
  
  
  
  
  22.             if delay_list:
    23. 
  
  
  
  
  23.                 delay = delay_list.pop()
    24. 
  
  
  
  
  24.                 task = asyncio.create_task(req(delay))
    25. 
  
  
  
  
  25.                 task_list.append(task)
    26. 
  
  
  
  
  26.             else:
    27. 
  
  
  
  
  27.                 break
    28. 
  
  
  
  
  28.         task_list = [task for task in task_list if not task.done()]
    29. 
  
  
  
  
  29.         await asyncio.sleep(1)
    30. 
  
  
  
  
  30.     end = time.time()
    31. 
  
  
  
  
  31.     print(f'一共耗时：{end - start}')
    32. 
  
  
  
  
  32. 
  
  
  
  
  33. asyncio.run(main())
    34.

运行效果如下图所示：

总共耗时大概28秒左右。比串行需要的58秒快了一半，但比全部同时并发多了一倍。

使用 Semaphore

asyncio 实际上自带了一个限制协程数量的类，叫做Semaphore。我们只需要初始化它，传入最大允许的协程数量，然后就可以通过上下文管理器来使用。我们看一下代码：

 
 
 
 
    
  
  
  
  
  1. import asyncio
    2. 
  
  
  
  
  2. import httpx
    3. 
  
  
  
  
  3. import time
    4. 
  
  
  
  
  4. 
  
  
  
  
  5. 
  
  
  
  
  6. async def req(delay, sem):
    7. 
  
  
  
  
  7.     print(f'请求一个延迟为{delay}秒的接口')
    8. 
  
  
  
  
  8.     async with sem:
    9. 
  
  
  
  
  9.         async with httpx.AsyncClient(timeout=20) as client:
    10. 
  
  
  
  
  10.             resp = await client.get(f'http://127.0.0.1:8000/sleep/{delay}')
    11. 
  
  
  
  
  11.             result = resp.json()
    12. 
  
  
  
  
  12.             print(result)
    13. 
  
  
  
  
  13. 
  
  
  
  
  14. 
  
  
  
  
  15. async def main():
    16. 
  
  
  
  
  16.     start = time.time()
    17. 
  
  
  
  
  17.     delay_list = [3, 6, 1, 8, 2, 4, 5, 2, 7, 3, 9, 8]
    18. 
  
  
  
  
  18.     task_list = []
    19. 
  
  
  
  
  19.     sem = asyncio.Semaphore(3)
    20. 
  
  
  
  
  20.     for delay in delay_list:
    21. 
  
  
  
  
  21.         task = asyncio.create_task(req(delay, sem))
    22. 
  
  
  
  
  22.         task_list.append(task)
    23. 
  
  
  
  
  23.     await asyncio.gather(*task_list)
    24. 
  
  
  
  
  24. 
  
  
  
  
  25.     end = time.time()
    26. 
  
  
  
  
  26.     print(f'一共耗时：{end - start}')
    27. 
  
  
  
  
  27. 
  
  
  
  
  28. asyncio.run(main())
    29.

运行效果如下图所示：

耗时为22秒，比第一个方案更快。

我们来看看Semaphore的用法，它的格式为：

 
 
 
 
    
  
  
  
  
  1. sem = asyncio.Semaphore(同时运行的协程数量)
    2. 
  
  
  
  
  2. 
  
  
  
  
  3. async def func(sem):
    4. 
  
  
  
  
  4.     async with sem:
    5. 
  
  
  
  
  5.         这里是并发执行的代码
    6. 
  
  
  
  
  6. 
  
  
  
  
  7. task_list = []
    8. 
  
  
  
  
  8. for _ in range(总共需要执行的任务数):
    9. 
  
  
  
  
  9.     task = asyncio.create_task(func(sem))
    10. 
  
  
  
  
  10.     task_list.append(task)
    11. 
  
  
  
  
  11. await asyncio.gather(*task_list)
    12.

当我们要限制一个协程的并发数的时候，可以在调用协程之前，先初始化一个Semaphore对象。然后把这个对象传到需要限制并发的协程里面，在协程里面，使用异步上下文管理器包住你的正式代码：

 
 
 
 
    
  
  
  
  
  1. async with sem:
    2. 
  
  
  
  
  2.     正式代码
    3.

这样一来，如果并发数没有达到限制，那么async with sem会瞬间执行完成，进入里面的正式代码中。如果并发数已经达到了限制，那么其他的协程会阻塞在async with sem这个地方，直到正在运行的某个协程完成了，退出了，才会放行一个新的协程去替换掉这个已经完成的协程。

这个写法其实跟多线程的加锁很像。只不过锁是确保同一个时间只有一个线程在运行，而Semaphore可以人为指定能有多少个协程同时运行。

如何限制1分钟内能够运行的协程数

可能同学看了上面的例子以后，只知道如何限制同时运行的协程数。但是怎么限制在一段时间里同时运行的协程数呢?

其实非常简单，在并发的协程里面加个 asyncio.sleep 就可以了。例如上面的例子，我想限制每分钟只能有3个协程，那么可以把代码改为：

 
 
 
 
    
  
  
  
  
  1. async def req(delay, sem):
    2. 
  
  
  
  
  2.     print(f'请求一个延迟为{delay}秒的接口')
    3. 
  
  
  
  
  3.     async with sem:
    4. 
  
  
  
  
  4.         async with httpx.AsyncClient(timeout=20) as client:
    5. 
  
  
  
  
  5.             resp = await client.get(f'http://127.0.0.1:8000/sleep/{delay}')
    6. 
  
  
  
  
  6.             result = resp.json()
    7. 
  
  
  
  
  7.             print(result)
    8. 
  
  
  
  
  8.     await asyncio.sleep(60)
    9.

总结

如果大家要限制协程的并发数，那么最简单的办法就是使用asyncio.Semaphore。但需要注意的是，只能在启动协程之前初始化它，然后传给协程。要确保所有并发协程拿到的是同一个Semaphore对象。

当然，你的程序里面，可能有多个不同的部分，有些部分限制并发数为 a，有些部分限制并发数为 b。那么你可以初始化多个Semaphore对象，分别传给不同的协程。

本文转载自微信公众号「未闻Code」，可以通过以下二维码关注。转载本文请联系未闻Code公众号。

当前文章：一日一技：使用Asyncio如何限制协程的并发数
文章起源：http://www.shufengxianlan.com/qtweb/news43/320893.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

手机网站建设知识

分类信息网

成都小程序开发公司　　 　新都四新网站建设　　 　网络营销公司　　 　成都网络推广　　 　香港云主机　　 　云南护栏打桩机　　 　百度推广公司　　 　广安网站建设　　 　营山网站建设　　 　达州主机托管　　 　西部信息服务器托管　　 　模板商城网站　　 　成都手机网站报价　　 　上海济桢教育　　 　重庆外贸网站建设　　 　香港虚拟主机空间　　 　成都网站建设　　 　服务器租用　　 　四川绵阳机房　　 　眉山电信机房