python异步编程模块asyncio学习

Posted 2021-11-07 c-x-a

尽管asyncio应用通常作为单线程运行，不过仍被构建为并发应用。由于I/O以及其他外部事件的延迟和中断，每个协程或任务可能按一种不可预知的顺序执行。为了支持安全的并发执行，asyncio包含了threading和multiprocessing模块中的一些底层原语的实现。

锁（LOCK）

锁可以用来保护对一个共享资源的访问。只有锁的持有者可以使用这个资源。如果有多个请求要的到这个锁，那么其将会阻塞，以保证一次只有一个持有者。
看一个锁的例子：

import asyncio
from functools import partial


def unlock(lock):
    print("callback释放锁")
    lock.release()


async def coro1(lock):
    print("并行中，coro1等待锁")
    async with lock:
        print("coro1被锁了")
    print("coro1的锁释放了")


async def coro2(lock):
    print("并行中，coro2等待锁")
    await lock.acquire()
    print(f"当前是否被锁", lock.locked())
    try:
        print("coro2被锁了")
    finally:
        lock.release()
        print("coro2的锁释放了")


async def coro3(lock):
    print("并行中，coro3等待锁")
    try:
        print("coro3没有加锁加试图释放")
        lock.release()
    except RuntimeError as e:
        print("触发RuntimeError的错误")


async def main(loop):
    # 创建一个锁
    lock = asyncio.Lock()
    loop.call_later(0.1, partial(unlock, lock))
    print("等待协程")
    await asyncio.wait([coro1(lock), coro2(lock), coro3(lock)])


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main(loop))
    finally:
        loop.close()

输出结果：

通过acquire加锁
当前是否被锁 True
等待协程
并行中，coro2等待锁
并行中，coro1等待锁
callback释放锁
coro2被锁了
coro2的锁释放了
coro1被锁了
coro1的锁释放了

通过上面的代码以及结果可以得出以下结论：

• lock.acquire()需要使用await。
• lock.release()不需要加await。
• 在加锁之前coro1和coro2 ，coro3是并发执行的。
• 锁有两种使用方式和像coro1一样通过async with 异步上下文关键字进行锁定，还可以通过coro2那种通过await方式使用acquire加锁，结束的时候使用release释放锁。

• 如果没有使用acquire进行加锁，就试图使用release去释放，将触发RuntimeError的异常，像coro3协程一样。

  事件(Event)

  asyncio.Event基于threading.Event。允许多个消费者等待某个事件发生，而不必寻找一个特定值与通知关联。

import asyncio
import functools


def callback(event):
    print('callback中设置event')
    event.set()


async def coro1(name, event):
    print(f'{name}等待事件')
    await event.wait()
    print(f'{name}触发')


async def coro2(name, event):
    print(f'{name}等待事件')
    await event.wait()
    print(f'{name}触发')


async def main(loop):
    event = asyncio.Event()
    print(f'当前事件状态: {event.is_set()}')
    loop.call_later(
        0.1, functools.partial(callback, event)
    )
    await asyncio.wait([coro1('coro1', event), coro2('coro2', event)])
    print(f'当前事件状态: {event.is_set()}')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))
    loop.close()

输出

当前事件状态: False
coro2等待事件
coro1等待事件
callback中设置event
coro2触发
coro1触发
当前事件状态: True

emmmm.。。。。。发现好像和lock也没啥区别。其实区别的话就是一旦触发了事件，coro1和coro2协程就会立即启动，不需要得到事件对象上的唯一的锁了。

条件(condition)

Condition的做法与Event类似，只不过不是通知所有的协程等待的协程，被唤醒的等待协程的数目由notify()的一个参数控制。

import asyncio


async def consumer(cond, name, second):
    await asyncio.sleep(second)
    async with cond:
        await cond.wait()
        print(f'{name}:资源可供消费者使用')


async def producer(cond):
    await asyncio.sleep(2)
    for n in range(1, 3):
        async with cond:
            print(f'唤醒消费者 {n}')
            cond.notify(n=n)
        await asyncio.sleep(0.1)


async def producer2(cond):
    await asyncio.sleep(2)
    with await cond:
        print('让资源变的可用')
        cond.notify_all()


async def main(loop):
    condition = asyncio.Condition()

    task = loop.create_task(producer(condition))
    consumers = [consumer(condition, name, index)
                 for index, name in enumerate(('c1', 'c2'))]
    await asyncio.wait(consumers)
    task.cancel()

    task = loop.create_task(producer2(condition))
    consumers = [consumer(condition, name, index)
                 for index, name in enumerate(('c1', 'c2'))]
    await asyncio.wait(consumers)
    task.cancel()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))
    loop.close()

输出内容

唤醒消费者 1
c1:资源可供消费者使用
唤醒消费者 2
c2:资源可供消费者使用
让资源变的可用
c1:资源可供消费者使用
c2:资源可供消费者使用

对上面的代码做简单的分析
使用notify方法挨个通知单个消费者
使用notify_all方法一次性的通知全部消费者
由于producer和producer2是异步的函数，所以不能使用之前call_later方法，需要用create_task把它创建成一个任务,或者asyncio.ensure_future().

队列(Queue)

asyncio.Queue为协程提供了一个先进先出的数据结构，这与线程queue.Queue或进程的multiprocess,Queue很类似。
这里直接上一个aiohtpp爬虫使用的例子

import aiohttp
import asyncio
import async_timeout
from urllib.parse import urljoin, urldefrag

root_url = "http://python.org/"
crawled_urls, url_hub = [], [root_url, f"{root_url}/sitemap.xml", f"{root_url}/robots.txt"]
headers = {
    'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_2) AppleWebKit/537.36 (Khtml, like Gecko) Chrome/72.0.3626.121 Safari/537.36'}


async def get_body(url):
    async with aiohttp.ClientSession() as session:
        try:
            with async_timeout.timeout(10):
                async with session.get(url, headers=headers) as response:
                    if response.status == 200:
                        html = await response.text()
                        return {'error': '', 'html': html}
                    else:
                        return {'error': response.status, 'html': ''}
        except Exception as err:
            return {'error': err, 'html': ''}


async def handle_task(task_id, work_queue):
    while not work_queue.empty():  # 如果队列不为空
        queue_url = await work_queue.get()  # 从队列中取出一个元素
        if not queue_url in crawled_urls:
            crawled_urls.append(queue_url)  # crawled_urls可以做一个去重操作
            body = await get_body(queue_url)
            if not body['error']:
                for new_url in get_urls(body['html']):
                    if root_url in new_url and not new_url in crawled_urls:
                        work_queue.put_nowait(new_url)
            else:
                print(f"Error: {body['error']} - {queue_url}")


def remove_fragment(url):
    pure_url, frag = urldefrag(url)
    return pure_url


def get_urls(html):
    new_urls = [url.split('"')[0] for url in str(html).replace("'", '"').split('href="')[1:]]
    return [urljoin(root_url, remove_fragment(new_url)) for new_url in new_urls]


if __name__ == "__main__":
    q = asyncio.Queue()  # 定义一个队列
    [q.put_nowait(url) for url in url_hub]  # 通过put_nowait方法循环往队列添加元素
    loop = asyncio.get_event_loop()
    tasks = [handle_task(task_id, q) for task_id in range(3)]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()
    for u in crawled_urls:
        print(u)
    print('-' * 30)
    print(len(crawled_urls))

代码中对关键的部分做了注释总结下Queue的内容：
1.通过asyncio.Queue():定义一个asyncio队列。
2.通过put_nowait可以向队列添加元素。
3.通过empty判断队列中的元素是否为空
4.get方法取出每个元素，需要注意的是要使用await。

信号量(Semaphore)

通过并发量可以去控制协程的并发数，爬虫操作使用中使用该方法减小并发量，可以减少对服务器的压力。Semaphore运作机制可以用停车场停车来比喻，一个停车场5个停车位，第一次5辆车可以都停下，我们知道正常情况下，不能在进入第6辆了，需要有一辆开走，然后才能再来一辆，当然如果有2辆开走，那么可以再同时进来2辆，一次类推我们就知道了，整个过程的关键点是，在车数足够多的时候，整个停车场最多只能放5辆车。下面我们在看个代码进一步做一个了解。

import aiohttp
import asyncio

URL = "http://www.baidu.com"
sem = asyncio.Semaphore(5)


async def branch():
    async with sem:
        await fetch()
        await asyncio.sleep(2)


async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(URL) as req:
            status = req.status
            print("状态码", status)


async def run():
    await branch()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        tasks=[asyncio.ensure_future(run()) for _ in range(21)]
        loop.run_until_complete(asyncio.wait(tasks))
    finally:
        loop.close()

上面代码是一个并发访问百度然后获取状态码的一个简单的例子，并发次数为20次，然后通过asyncio.Semaphore(5)指定并发量为5,通过async with sem做一个限制，然后fetch协程是整个爬虫的逻辑代码，运行上面的代码可以发现每隔2s输出5个请求结果。
其他地方的使用第一章内容讲的很详细了这里就不详细说了。