从零开始学asyncio(下)

import requests


def fetch(url):
    response = requests.get(url)
    # 以下省略一万字

import asyncio
import requests


async def fetch(url):
    loop = asyncio.get_running_loop()
    asyncIO = loop.run_in_executor(None, requests.get, url)
    response = await asyncIO
    # 继续省略一万字

import asyncio


class Lock:

    def __init__(self):
        self._locked = False

    async def __aenter__(self):
        while self._locked:
            await asyncio.sleep(0)
        self._locked = True
        return self

    async def __aexit__(self, exc_type, exc, tb):
        self._locked = False


lock = Lock()


async def coro():
    async with lock:
        pass


asyncio.run(coro())

import asyncio


class Xrange:

    def __init__(self, lower, upper):
        self._num = lower-1
        self.upper = upper

    def __aiter__(self):
        return self

    async def __anext__(self):
        self._num += 1
        if self._num < self.upper:
            return self._num
        raise StopAsyncIteration


async def coro():
    async for i in Xrange(1, 10):
        print(i)


asyncio.run(coro())

def close(self):
    try:
        self.throw(GeneratorExit)
    except (GeneratorExit, StopIteration):
        pass
    else:
        raise RuntimeError("generator ignored GeneratorExit")

# 要把协程封装为task对象,可以使用以下两个函数:

# 这个函数必须在有事件循环运行时调用,不然报错
# 一般来说,在协程中调用就行
asyncio.create_task(coro)

# 这个函数不光可以接收协程对象,只要是可等待对象都可以
# 如果没有事件循环, 它就会调用get_event_loop获取
asyncio.ensure_future(aw)


# 或者在获取到当前事件循环的前提下
loop.create_task()


# 如果要创建future对象,可以使用以下方法:

asyncio.Future()

# 或者在获取到当前事件循环的前提下
loop.create_future()

import asyncio


loop = asyncio.get_event_loop()


async def coro():
    fut = loop.create_future()

    def cb(fut):
        print(‘这是一个回调函数‘)
    fut.add_done_callback(cb)
    # 在一秒之后取消fut
    loop.call_later(1, fut.cancel)
    try:
        await fut
    except Exception as e:
        print(‘异常:‘, type(e))
    print(‘future对象:‘, fut)


loop.run_until_complete(coro())

# 如果没有正在运行的事件循环, 下面这个函数会报错
# 因此, 只能在协程或者相关的回调中调用这个函数
asyncio.get_running_loop()
# 获取事件循环, 没有就会新建一个
asyncio.get_event_loop()
# 新建一个事件循环
asyncio.new_event_loop()
# 将loop绑定到当前的线程中
asyncio.set_event_loop(loop)

import asyncio


loop = asyncio.get_event_loop()


async def coro():
    fut = loop.create_future()
    loop.call_later(1, fut.set_result, None)
    await fut
    print(‘the end‘)


asyncio.run(coro())

import asyncio
import os
import random
import re
import time

import aiofiles
import aiohttp


class Producer:

    def __init__(self, session, queue):
        self.session = session
        self.regex = re.compile(
            ‘‘‘<div id="comic">\n<img src="(//imgs.xkcd.com/comics/[a-z0-9_]+\.(jpg|png))"‘‘‘)
        self.q = queue
        self.urls = self.url_factory()

    async def start(self):
        workers = [self.worker() for i in range(10)]
        await asyncio.wait(workers)
        self.q.put_nowait(‘all tasks are done‘)

    async def worker(self):
        while self.urls:
            url = self.urls.pop()
            html = await self.fetch(url)
            await self.pares_html(html)

    def url_factory(self, n=100):
        # 返回一个含有n个url的集合
        urls = set()
        while len(urls) < n:
            urls.add(‘https://xkcd.com/{}/‘.format(random.randrange(1, 2245)))
        return urls

    async def fetch(self, url):
        async with self.session.get(url) as response:
            assert response.status == 200
            return await response.text()

    async def pares_html(self, html):
        res = re.search(self.regex, html)
        if res is None:
            return
        img_url = ‘https:‘+res.group(1)
        await self.q.put(img_url)


class Consumer:

    def __init__(self, session, queue):
        self.session = session
        self.q = queue
        self.save_folder = os.path.join(os.path.dirname(__file__), ‘webcomics‘)
        try:
            os.mkdir(self.save_folder)
        except FileExistsError:
            pass

    async def start(self):
        workers = [self.worker() for i in range(10)]
        await asyncio.wait(workers)

    async def worker(self):
        while 1:
            url = await self.q.get()
            if url == ‘all tasks are done‘:
                break
            await self.download(url)
        self._on_worker_done()

    def _on_worker_done(self):
        self.q.put_nowait(‘all tasks are done‘)

    async def download(self, url):
        filename = os.path.split(url)[-1]
        save_path = os.path.join(self.save_folder, filename)
        if os.path.exists(save_path):
            return
        async with self.session.get(url) as response:
            assert response.status == 200
            content = await response.read()
            async with aiofiles.open(save_path, ‘wb‘) as f:
                await f.write(content)


async def main():
    queue = asyncio.Queue(maxsize=100)
    async with aiohttp.ClientSession() as session:
        producer = Producer(session, queue)
        consumer = Consumer(session, queue)
        tasks = asyncio.gather(producer.start(), consumer.start())
        await tasks


if __name__ == ‘__main__‘:
    asyncio.run(main())

async def start(self):
    workers = [asyncio.Task(self.worker()) for i in range(10)]
    await self.q.join()
    for worker in workers:
        worker.cancel()

async def worker(self):
    while 1:
        url = await self.q.get()
        await self.download(url)
        self.q.task_done()