【原】python并发编程：Python异步IO实现并发爬虫

攻城狮成长日志 2023-10-13 发布于广东

展开全文

往期文章：

协程内容的介绍

❝
上图的上面是单线程爬虫 cpu的执行情况，可以发现，经常因为等待IO而影响CPU的执行效率。
上图的下面是协程，协程主要是在单线程内实现的，以爬虫为例，协程先是让cpu爬取第一个url的内容，等待IO的时候，它又让CPU爬取第二个url的内容，当第二个任务等待IO的时候，它又让CPU爬取第三个url的内容，然后第三个任务等待IO, 它又循环回来，执行第一个任务，就这样返回循环。所以，协程就是大循环。
❞

asyncio使用

import asyncio

# 获取事件循环
loop = asyncio.get_event_loop()

# 定义协程
async def myfunc(url):
    await get_url(url)

# 创建task列表
tasks = [loop.create_task(myfunc(url)) for url in urls]

# 执行爬虫事件列表
loop.run_until_complete(asyncio.wait(tasks))

❝
注意：
要用在异步IO编程中, 依赖的库必须支持异步IO特性
爬虫引用中：requests 不支持异步, 需要用 aiohttp
❞

代码演示

import aiohttp
import asyncio
from loguru import logger
from  cnblogs_spider import urls
import time

async def async_craw(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            result = await resp.text()
            logger.info("craw url {},{}".format(url,len(result)))


loop = asyncio.get_event_loop()
# 定义超级循环
tasks = [ loop.create_task(async_craw(url))  for url in urls]


start = time.time()
loop.run_until_complete(asyncio.wait(tasks))
end = time.time()
logger.info("use time {}秒".format(end-start))

执行结果如下:

信号量

信号量（Semaphore）是一种并发控制机制，用于保护共享资源的访问。它是由计数器和等待队列组成的对象，用于控制对共享资源的访问权限。是一个同步对象，用于保持在0至指定最大值之间的一个计数值。

当线程完成一次对该semaphore对象的等待（wait）时，该计数值减一；
当线程完成一次对semaphore对象的释放（release）时，计数值加一。
当计数值为0，则线程等待该semaphore对象不再能成功直至该semaphore对象变成signaled状态
semaphore对象的计数值大于0，为signaled状态；计数值等于0，为nonsignaled状态.

「信号量是用来控制并发度的。」

主要有两种实现方式:

方式一:

sem = asyncio.Semaphore(10)

# ... later
async with sem:
    # work with shared resource

方式二：

sem = asyncio.Semaphore(10)

# ... later
await sem.acquire()
try:
    # work with shared resource
finally:
    sem.release()

用信号量控制协程数进行爬虫

import aiohttp
import asyncio
from loguru import logger
from  cnblogs_spider import urls
import time



# 加入信号量，控制并发度
semaphore = asyncio.Semaphore(10)

async def async_craw(url):
    async with semaphore:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                result = await resp.text()
                logger.info("craw url {},{}".format(url,len(result)))


loop = asyncio.get_event_loop()
# 定义超级循环
tasks = [ loop.create_task(async_craw(url))  for url in urls]


start = time.time()
loop.run_until_complete(asyncio.wait(tasks))
end = time.time()
logger.info("use time {}秒".format(end-start))