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

也想出现在这里？点击联系我~

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

提出问题

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

1. importasyncio
2. importhttpx
3. importtime
6. asyncdefreq(delay):
7. print(f'请求一个延迟为{delay}秒的接口')
8. asyncwithhttpx.AsyncClient(timeout=20)asclient:
9. resp=awaitclient.get(f'http://127.0.0.1:8000/sleep/{delay}')
10. result=resp.json()
11. print(result)
14. asyncdefmain():
15. start=time.time()
16. delay_list=[3,6,1,8,2,4,5,2,7,3,9,8]
17. task_list=[]
18. fordelayindelay_list:
19. task=asyncio.create_task(req(delay))
20. task_list.append(task)
21. awaitasyncio.gather(*task_list)
22. end=time.time()
23. print(f'一共耗时：{end-start}')
25. asyncio.run(main())

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

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

限制协程任务数

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

1. importasyncio
2. importhttpx
3. importtime
6. asyncdefreq(delay):
7. print(f'请求一个延迟为{delay}秒的接口')
8. asyncwithhttpx.AsyncClient(timeout=20)asclient:
9. resp=awaitclient.get(f'http://127.0.0.1:8000/sleep/{delay}')
10. result=resp.json()
11. print(result)
14. asyncdefmain():
15. start=time.time()
16. delay_list=[3,6,1,8,2,4,5,2,7,3,9,8]
17. task_list=[]
18. whileTrue:
19. ifnotdelay_listandnottask_list:
20. break
21. whilelen(task_list)<3:
22. ifdelay_list:
23. delay=delay_list.pop()
24. task=asyncio.create_task(req(delay))
25. task_list.append(task)
26. else:
27. break
28. task_list=[taskfortaskintask_listifnottask.done()]
29. awaitasyncio.sleep(1)
30. end=time.time()
31. print(f'一共耗时：{end-start}')
33. asyncio.run(main())

运行效果如下图所示：

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

使用 Semaphore

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

1. importasyncio
2. importhttpx
3. importtime
6. asyncdefreq(delay,sem):
7. print(f'请求一个延迟为{delay}秒的接口')
8. asyncwithsem:
9. asyncwithhttpx.AsyncClient(timeout=20)asclient:
10. resp=awaitclient.get(f'http://127.0.0.1:8000/sleep/{delay}')
11. result=resp.json()
12. print(result)
15. asyncdefmain():
16. start=time.time()
17. delay_list=[3,6,1,8,2,4,5,2,7,3,9,8]
18. task_list=[]
19. sem=asyncio.Semaphore(3)
20. fordelayindelay_list:
21. task=asyncio.create_task(req(delay,sem))
22. task_list.append(task)
23. awaitasyncio.gather(*task_list)
25. end=time.time()
26. print(f'一共耗时：{end-start}')
28. asyncio.run(main())

运行效果如下图所示：

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

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

1. sem=asyncio.Semaphore(同时运行的协程数量)
3. asyncdeffunc(sem):
4. asyncwithsem:
5. 这里是并发执行的代码
7. task_list=[]
8. for_inrange(总共需要执行的任务数):
9. task=asyncio.create_task(func(sem))
10. task_list.append(task)
11. awaitasyncio.gather(*task_list)

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

1. asyncwithsem:
2. 正式代码

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

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

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

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

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

1. asyncdefreq(delay,sem):
2. print(f'请求一个延迟为{delay}秒的接口')
3. asyncwithsem:
4. asyncwithhttpx.AsyncClient(timeout=20)asclient:
5. resp=awaitclient.get(f'http://127.0.0.1:8000/sleep/{delay}')
6. result=resp.json()
7. print(result)
8. awaitasyncio.sleep(60)

总结

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

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

原文链接：https://mp.weixin.qq.com/s/zs319rzFDE5y7v4egbZOUQ