使用Python进行并发编程-asyncio篇(二)

发表于 | 分类于 | 阅读次数:

我们今天继续深入学习asyncio。

同步机制

asyncio模块包含多种同步机制，每个原语的解释可以看线程篇，这些原语的用法上和线程/进程有一些区别。

Semaphore（信号量）

并发的去爬取显然可以让爬虫工作显得更有效率，但是我们应该把抓取做的无害，这样既可以保证我们不容易发现，也不会对被爬的网站造成一些额外的压力。
在这里吐槽下，豆瓣现在几乎成了爬虫练手专用网站，我个人也不知道为啥？欢迎留言告诉我。难道是豆瓣一直秉承尊重用户的原则不轻易对用户才去封禁策略，造成大家觉得豆瓣最适合入门么？BTW，我每天在后台都能看到几十万次无效的抓取，也就是抓取程序写的有问题，但还在不停地请求着…
好吧回到正题，比如我现在要抓取http://httpbin.org/get?a=X这样的页面，X为1-10000的数字，一次性的产生1w次请求显然很快就会被封掉。那么我们可以用Semaphore控制同时的并发量(例子中为了演示，X为0-11)：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import aiohttp  
import asyncio  
  
NUMBERS = range(12)  
URL = 'http://httpbin.org/get?a={}'  
sema = asyncio.Semaphore(3)  
  
  
async def fetch_async(a):  
    async with aiohttp.request('GET', URL.format(a)) as r:  
        data = await r.json()  
    return data['args']['a']  
  
  
async def print_result(a):  
    with (await sema):  
        r = await fetch_async(a)  
        print('fetch({}) = {}'.format(a, r))  
  
  
loop = asyncio.get_event_loop()  
f = asyncio.wait([print_result(num) for num in NUMBERS])  
loop.run_until_complete(f)

在运行的时候可以感受到并发受到了信号量的限制，基本保持在同时处理三个请求的标准。

Lock（锁）

看下面的例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

❯ cat lock.py  
import asyncio  
import functools  
  
  
def unlock(lock):  
    print('callback releasing lock')  
    lock.release()  
  
  
async def test(locker, lock):  
    print('{} waiting for the lock'.format(locker))  
    with await lock:  
        print('{} acquired lock'.format(locker))  
    print('{} released lock'.format(locker))  
  
  
async def main(loop):  
    lock = asyncio.Lock()  
    await lock.acquire()  
    loop.call_later(0.1, functools.partial(unlock, lock))  
    await asyncio.wait([test('l1', lock), test('l2', lock)])  
  
loop = asyncio.get_event_loop()  
loop.run_until_complete(main(loop))  
loop.close()

这个例子中我们首先使用acquire加锁，通过call_later方法添加一个0.1秒后释放锁的函数。看一下调用：

1
2
3
4
5
6
7
8
9

❯ python3 lock.py  
l1 waiting for the lock  
l2 waiting for the lock  
callback releasing lock  
l1 acquired lock  
l1 released lock  
l2 acquired lock  
l2 released lock

Condition（条件）

我们根据线程篇Condition的例子，改成一下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

import asyncio  
import functools  
  
  
async def consumer(cond, name, second):  
    await asyncio.sleep(second)  
    with await cond:  
        await cond.wait()  
        print('{}: Resource is available to consumer'.format(name))  
  
  
  
async def producer(cond):  
    await asyncio.sleep(2)  
    for n in range(1, 3):  
        with await cond:  
            print('notifying consumer {}'.format(n))  
            cond.notify(n=n)  
        await asyncio.sleep(0.1)  
  
async def producer2(cond):  
    await asyncio.sleep(2)  
    with await cond:  
        print('Making resource available')  
        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()  
  
  
loop = asyncio.get_event_loop()  
loop.run_until_complete(main(loop))  
loop.close()

这次演示了2种通知的方式：

  1. 使用notify方法挨个通知单个消费者
  2. 使用notify_all方法一次性的通知全部消费者
    由于producer和producer2是异步的函数，所以不能使用之前call_later方法，需要用create_task把它创建成一个任务（Task）。但是最后记得要把任务取消掉。
    执行以下看看效果：
1
2
3
4
5
6
7
8
9

❯ python3 condition.py  
notifying consumer 1  
c1: Resource is available to consumer  
notifying consumer 2  
c2: Resource is available to consumer  
Making resource available  
c1: Resource is available to consumer  
c2: Resource is available to consumer

Event（事件）

模仿锁的例子实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import asyncio  
import functools  
  
  
def set_event(event):  
    print('setting event in callback')  
    event.set()  
  
  
async def test(name, event):  
    print('{} waiting for event'.format(name))  
    await event.wait()  
    print('{} triggered'.format(name))  
  
  
async def main(loop):  
    event = asyncio.Event()  
    print('event start state: {}'.format(event.is_set()))  
    loop.call_later(  
        0.1, functools.partial(set_event, event)  
    )  
    await asyncio.wait([test('e1', event), test('e2', event)])  
    print('event end state: {}'.format(event.is_set()))  
  
loop = asyncio.get_event_loop()  
loop.run_until_complete(main(loop))  
loop.close()

看起来也确实和锁的意思很像，不同的是，事件被触发时，2个消费者不用获取锁就要尽快的执行下去了。

Queue

在asyncio官网上已经举例了2个很好的队列例子了，这文就不重复了。asyncio同样支持LifoQueue和PriorityQueue，我们体验下aiohttp+优先级队列的用法吧：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

import asyncio  
import random  
import aiohttp  
  
NUMBERS = random.sample(range(100), 7)  
URL = 'http://httpbin.org/get?a={}'  
sema = asyncio.Semaphore(3)  
  
  
async def fetch_async(a):  
    async with aiohttp.request('GET', URL.format(a)) as r:  
        data = await r.json()  
    return data['args']['a']  
  
  
async def collect_result(a):  
    with (await sema):  
        return await fetch_async(a)  
  
  
async def produce(queue):  
    for num in NUMBERS:  
        print('producing {}'.format(num))  
        item = (num, num)  
        await queue.put(item)  
  
  
async def consume(queue):  
    while 1:  
        item = await queue.get()  
        num = item[0]  
        rs = await collect_result(num)  
        print('consuming {}...'.format(rs))  
        queue.task_done()  
  
  
async def run():  
    queue = asyncio.PriorityQueue()  
    consumer = asyncio.ensure_future(consume(queue))  
    await produce(queue)  
    await queue.join()  
    consumer.cancel()  
  
  
loop = asyncio.get_event_loop()  
loop.run_until_complete(run())  
loop.close()

看到使用了新的ensure_future方法，其实它和之前说的create_task意思差不多，都是为了把一个异步的函数变成一个协程的Task。它们的区别是：

  1. create_task是AbstractEventLoop的抽象方法，不同的loop可以实现不同的创建Task方法，这里用的是BaseEventLoop的实现。
  2. ensure_future是asyncio封装好的创建Task的函数，它还支持一些参数，甚至指定loop。一般应该使用它，除非用到后面提到的uvloop这个第三方库。
    这个例子中，首先我们从0-99中随机取出7个数字，放入优先级队列，看看消费者是不是按照从小到大的顺序执行的呢？
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

❯ python3 prioqueue.py  
producing 6  
producing 4  
producing 22  
producing 48  
producing 9  
producing 90  
producing 40  
consuming 4...  
consuming 6...  
consuming 9...  
consuming 22...  
consuming 40...  
consuming 48...  
consuming 90...

确实是这样的。
说到这里，我们稍微偏个题，看看Task是什么？

深入Task

Task类用来管理协同程序运行的状态。根据源码，我保留核心，实现一个简单的Task类帮助大家理解：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import asyncio  
  
  
class Task(asyncio.futures.Future):  
    def __init__(self, gen, *,loop):  
        super().__init__(loop=loop)  
        self._gen = gen  
        self._loop.call_soon(self._step)  
  
    def _step(self, val=None, exc=None):  
        try:  
            if exc:  
                f = self._gen.throw(exc)  
            else:  
                f = self._gen.send(val)  
        except StopIteration as e:  
            self.set_result(e.value)  
        except Exception as e:  
            self.set_exception(e)  
        else:  
            f.add_done_callback(  
                 self._wakeup)  
  
    def _wakeup(self, fut):  
        try:  
            res = fut.result()  
        except Exception as e:  
            self._step(None, e)  
        else:  
            self._step(res, None)

如果_step方法没有让协程执行完成，就会添加回调，_wakeup又会继续执行_step… 直到协程程序完成，并set_result。
写个使用它的例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

async def foo():  
    await asyncio.sleep(2)  
    print('Hello Foo')  
  
  
async def bar():  
    await asyncio.sleep(1)  
    print('Hello Bar')  
  
  
loop = asyncio.get_event_loop()  
tasks = [Task(foo(), loop=loop),  
         loop.create_task(bar())]  
loop.run_until_complete(  
        asyncio.wait(tasks))  
loop.close()

第一个任务是用我们自己的Task创建的，第二个是用BaseEventLoop自带的create_task。
运行一下：

1
2
3
4

❯ python3 task.py  
Hello Bar  
Hello Foo

自定义的Task类和asyncio自带的是可以好好协作的。

深入事件循环

asyncio根据你的操作系统信息会帮你选择默认的事件循环类，在*nix下使用的类继承于BaseEventLoop，在上面已经提到了。和Task一样，我们剥离出一份最核心的实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

import asyncio  
from collections import deque  
  
  
def done_callback(fut):  
    fut._loop.stop()  
  
  
class Loop:  
    def __init__(self):  
        self._ready = deque()  
        self._stopping = False  
  
    def create_task(self, coro):  
        Task = asyncio.tasks.Task  
        task = Task(coro, loop=self)  
        return task  
  
    def run_until_complete(self, fut):  
        tasks = asyncio.tasks  
        # 获取任务  
        fut = tasks.ensure_future(  
                    fut, loop=self)  
        # 增加任务到self._ready  
        fut.add_done_callback(done_callback)  
        # 跑全部任务  
        self.run_forever()  
        # 从self._ready中移除  
        fut.remove_done_callback(done_callback)  
  
    def run_forever(self):  
        try:  
            while 1:  
                self._run_once()  
                if self._stopping:  
                    break  
        finally:  
            self._stopping = False  
  
    def call_soon(self, cb, *args):  
        self._ready.append((cb, args))  
  
    def _run_once(self):  
        ntodo = len(self._ready)  
        for i in range(ntodo):  
            t, a = self._ready.popleft()  
            t(*a)  
  
    def stop(self):  
        self._stopping = True  
  
    def close(self):  
        self._ready.clear()  
  
    def call_exception_handler(self, c):  
        pass  
  
    def get_debug(self):  
        return False

其中call_exception_handler和get_debug是必须存在的。
写个例子用一下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

async def foo():  
    print('Hello Foo')  
  
  
async def bar():  
    print('Hello Bar')  
  
loop = Loop()  
tasks = [loop.create_task(foo()),  
         loop.create_task(bar())]  
loop.run_until_complete(  
        asyncio.wait(tasks))  
loop.close()

执行：

1
2
3
4

❯ python3 loop.py  
Hello Foo  
Hello Bar

也可以和asyncio.wait正常协作了。
PS：本文全部代码可以在微信公众号文章代码库项目中找到。

版权声明：本文由 董伟明 原创，未经作者授权禁止任何微信公众号和向掘金(juejin.im)转载，技术博客转载采用 保留署名-非商业性使用-禁止演绎 4.0-国际许可协议
python