# 协程（微线程，纤程）
[TOC]
---
## 基础概念
多线程和多进程的模型虽然解决了并发问题，但是系统不能无上限地增加线程。由于系统切换线程的开销也很大，所以，一旦线程数量过多，CPU的时间就花在线程切换上了，真正运行代码的时间就少了，结果导致性能严重下降。由于我们要解决的问题是CPU高速执行能力和IO设备的龟速严重不匹配，多线程和多进程只是解决这一问题的一种方法。
另一种解决IO问题的方法是异步IO。当代码需要执行一个耗时的IO操作时，它只发出IO指令，并不等待IO结果，然后就去执行其他代码了。一段时间后，当IO返回结果时，再通知CPU进行处理。
### 同步IO模型和异步IO模型
```python
# 同步IO模型：
do_some_code()
f = open('/path/to/file', 'r')
r = f.read() # <== 线程停在此处等待IO操作结果
# IO操作完成后线程才能继续执行:
do_some_code(r)
# 异步io模型
loop = get_event_loop()
# 需要一个消息循环
while True:
# 主线程不断地重复“读取消息-处理消息”这一过程
event = loop.get_event()
process_event(event)
```
在“发出IO请求”到收到“IO完成”的这段时间里，同步IO模型下，主线程只能挂起，但异步IO模型下，主线程并没有休息，而是在消息循环中继续处理其他消息。这样，**在异步IO模型下，一个线程就可以同时处理多个IO请求，并且没有切换线程的操作。对于大多数IO密集型的应用程序，使用异步IO将大大提升系统的多任务处理能力。**
协程是实现异步IO的高级形式，又称微线程，纤程。英文名Coroutine。
子程序，或者称为函数，在所有语言中都是层级调用，比如A调用B，B在执行过程中又调用了C，C执行完毕返回，B执行完毕返回，最后是A执行完毕。
子程序调用总是一个入口，一次返回，调用顺序是明确的。而协程的调用和子程序不同。协程看上去也是子程序，但执行过程中，在子程序内部可中断，然后转而执行别的子程序，在适当的时候再返回来接着执行。
```
def A():
print('1')
print('2')
print('3')
def B():
print('x')
print('y')
print('z')
# 协程执行结果，可能如下
1
2
x
y
3
z
```
看起来A、B的执行有点像多线程，但协程的特点在于是一个线程执行，那和多线程比，协程有何优势？
主要行效率比多线程高很多，主要表现一下两点：
1. 不用切换线程，而是由程序自身控制，因此，没有线程切换的开销，和多线程比，线程数量越多，协程的性能优势就越明显。
2. 不需要多线程的锁机制，因为只有一个线程，也不存在同时写变量冲突，在协程中控制共享资源不加锁，只需要判断状态就好了。
因为协程是一个线程执行，那怎么利用多核CPU呢？最简单的方法是多进程+协程，既充分利用多核，又充分发挥协程的高效率，可获得极高的性能。
Python对协程的支持是通过generator实现的。
在generator中，我们不但可以通过for循环来迭代，还可以不断调用next()函数获取由yield语句返回的下一个值。Python的yield不但可以返回一个值，它还可以接收调用者发出的参数。
```python
import inspect
def consumer():
r = ''
while True:
# 3. 通过yield拿到消息n（最开始send进来的n为none，不返回只做启动用）
# yield关键字右边可以不需要加表达式（yield默认返回None）
n = yield r
if not n:
return
# 4. 拿到n之后进行处理
print('[CONSUMER] Consuming %s...' % n)
# 5. 处理完成，再下个循环又通过yield返回
r = '200 OK'
def produce(c):
# 1. 调用c.send(None)启动生成器；
# GEN_CREATED: 等待开始执行
print(inspect.getgeneratorstate(c))
c.send(None)
n = 0
while n < 3:
n = n + 1
print('[PRODUCER] Producing %s...' % n)
# 2. 一旦生产了东西，通过c.send(n)切换到consumer执行
r = c.send(n)
print(inspect.getgeneratorstate(c))
# 6. 得到consumer处理的结果，再通过下一个循环，继续生产下一条消息
print('[PRODUCER] Consumer return: %s' % r)
# 在close前，状态 都是 GEN_SUSPENDED # 在yield表达式处暂停
print(inspect.getgeneratorstate(c))
# 7. produce决定不生产了，通过c.close()关闭consumer，整个过程结束。
c.close()
# close后 状态为GEN_CLOSED # 执行结束
print(inspect.getgeneratorstate(c))
c = consumer()
produce(c)
```
> 整个流程无锁，由一个线程执行，`produce`和`consumer`协作完成任务，所以称为“协程”，而非线程的抢占式多任务。
### 协程生成器的基本行为
协程有四个状态，可以使用`inspect.getgeneratorstate(...)`函数确定：
GEN_CREATED # 等待开始执行
GEN_RUNNING # 解释器正在执行（只有在多线程应用中才能看到这个状态）
GEN_SUSPENDED # 在yield表达式处暂停
GEN_CLOSED # 执行结束
### 生成器api
1. `.send(value)`方法，生成器可以使用`.send(...)`方法发送数据，发送的数据会成为生成器函数中yield表达式的值。如上列中的n和r
2. `.throw(...)`方法，让调用方抛出异常，在生成器中处理
3. `.close()`方法，终止生成器
## asyncio
asyncio是Python 3.4版本引入的标准库，直接内置了对异步IO的支持，asyncio的编程模型就是一个消息循环。我们从asyncio模块中直接获取一个EventLoop的引用，然后把需要执行的协程扔到EventLoop中执行，就实现了异步IO。
### @asyncio.coroutine
用asyncio提供的@asyncio.coroutine可以把一个generator标记为coroutine类型，然后在coroutine内部用yield from调用另一个coroutine实现异步操作。
```python
import asyncio
import threading
# @asyncio.coroutine把一个generator标记为coroutine类型
@asyncio.coroutine
def baby(num):
print('baby %s sleep! (%s)' % (num,threading.currentThread()))
# 异步调用asyncio.sleep(2)生成器: 假设是一个耗时2秒的IO操作，在此期间，主线程并未等待，而是去执行EventLoop中其他可以执行的coroutine了，因此可以实现并发执行。
yield from asyncio.sleep(2)
print('baby %s week up! (%s)' % (num,threading.currentThread()))
# 获取EventLoop: 事件循环对象
loop = asyncio.get_event_loop()
tasks = [baby(1), baby(2), baby(3)]
# 把上面coroutine扔到EventLoop中执行
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
'''
baby 1 sleep! (<_MainThread(MainThread, started 29028)>)
baby 2 sleep! (<_MainThread(MainThread, started 29028)>)
baby 3 sleep! (<_MainThread(MainThread, started 29028)>)
# (暂停约2秒,并且是在同一线程里面,实现了并发)
baby 1 week up! (<_MainThread(MainThread, started 29028)>)
baby 2 week up! (<_MainThread(MainThread, started 29028)>)
baby 3 week up! (<_MainThread(MainThread, started 29028)>)
1. baby(1)执行到yield,线程不会等待asyncio.sleep(),而是直接中断并执行下一个消息循环baby(2)
2. baby(2)执行到yield,线程不会等待asyncio.sleep(),而是直接中断并执行下一个消息循环baby(3)
3. baby(3)执行到yield,线程不会等待asyncio.sleep(),而是直接中断并执行消息循环baby(1),至此所有操作都是以极快的时间完成的,花费了极少时间,此时三个baby同时都在睡眠,(asyncio.sleep)
4. 等待baby(1)睡眠完成,此时几乎同时其他baby也的睡眠也结束了,所以接着执行各个baby的打印wake up操作.结束整个消息循环,run_until_complete结束.
'''
```
**用asyncio的异步网络连接来获取sina、sohu和163的网站首页**
```python
import asyncio
@asyncio.coroutine
def wget(host):
print('wget %s...' % host)
connect = asyncio.open_connection(host, 80)
reader, writer = yield from connect
header = 'GET / HTTP/1.0\r\nHost: %s\r\n\r\n' % host
writer.write(header.encode('utf-8'))
# 刷新底层传输的写缓冲区。也就是把需要发送出去的数据，从缓冲区发送出去。没有手工刷新，asyncio为你自动刷新了。当执行到reader.readline()时，asyncio知道应该把发送缓冲区的数据发送出去了。
yield from writer.drain()
while True:
line = yield from reader.readline()
if line == b'\r\n':
break
print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
# Ignore the body, close the socket
writer.close()
loop = asyncio.get_event_loop()
tasks = [wget(host) for host in ['www.sina.com.cn', 'www.sohu.com', 'www.163.com']]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
```
### async/await
async和await是针对coroutine的新语法，要使用新的语法，只需要做两步简单的替换：
1. 把@asyncio.coroutine替换为async；
2. 把yield from替换为await。
使用async可以定义协程对象，使用await可以针对耗时的操作进行挂起，就像生成器里的yield一样，函数让出控制权。**协程遇到await，事件循环将会挂起该协程，执行别的协程，直到其他的协程也挂起或者执行完毕，再进行下一个协程的执行**。
```python
import asyncio
import threading
async def baby(num):
print('baby %s sleep! (%s)' % (num,threading.currentThread()))
await asyncio.sleep(1)
print('baby %s week up! (%s)' % (num,threading.currentThread()))
loop = asyncio.get_event_loop()
# ???? 执行完的顺序让人疑惑
tasks = [baby(2), baby(1), baby(3),baby(4),baby(5)]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
```
> tips: await 和 yield from 可以理解为 “不等了” （主线程是一个事件循环，执行到await，就“我不等了，您慢慢执行，我先走一步，好了再给我说”）
### 绑定回调
```python
import time
import asyncio
now = lambda : time.time()
async def do_some_work(x):
print('Waiting: ', x)
return 'Done after {}s'.format(x)
def callback(future):
print('Callback: ', future.result())
start = now()
coroutine = do_some_work(2)
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(coroutine)
task.add_done_callback(callback)
loop.run_until_complete(task)
print('TIME: ', now() - start)
```
**利用future对象回调别的函数**
?? future对象的特性,以下代码不太懂 ??
```python
import asyncio
import functools
def callback(future, n):
print('{}: future done: {}'.format(n, future.result()))
async def register_callbacks(all_done):
print('registering callbacks on future')
# 偏函数配合回调，all_done是future对象
all_done.add_done_callback(functools.partial(callback, n=1))
all_done.add_done_callback(functools.partial(callback, n=2))
async def main(all_done):
# 到此同步中断，异步执行回调函数注册
await register_callbacks(all_done)
print('setting result of future')
all_done.set_result('the result')
event_loop = asyncio.get_event_loop()
try:
all_done = asyncio.Future()
event_loop.run_until_complete(main(all_done))
finally:
event_loop.close()
'''
registering callbacks on future
setting result of future
1: future done: the result
2: future done: the result
'''
```
### 多线程与asyncio对比
**多线程**
```python
# sinner_thread.py
import threading
import itertools
import time
import sys
# 这个类定义一个可变对象，用于从外部控制线程
class Signal:
go = True
# 这个函数会在单独的线程中运行，signal 参数是前边定义的Signal类的实例
def spin(msg, signal):
write, flush = sys.stdout.write, sys.stdout.flush
# itertools.cycle 函数从指定的序列中反复不断地生成元素
for char in itertools.cycle('|/-\\'):
status = char + ' ' + msg
write(status)
flush()
write('\x08' * len(status)) # 使用退格符把光标移回行首
time.sleep(.1) # 每 0.1 秒刷新一次
if not signal.go: # 如果 go属性不是 True，退出循环
break
write(' ' * len(status) + '\x08' * len(status)) # 使用空格清除状态消息，把光标移回开头
def slow_function(): # 模拟耗时操作
# 假装等待I/O一段时间
time.sleep(20) # 调用sleep 会阻塞主线程，这么做事为了释放GIL，创建从属线程
return 42
# 这个函数设置从属线程，显示线程对象，运行耗时计算，最后杀死进程
def supervisor():
signal = Signal()
spinner = threading.Thread(target=spin,
args=('thinking!', signal))
print('spinner object:', spinner) # 显示线程对象 输出 spinner object: <Thread(Thread-1, initial)>
spinner.start() # 启动从属进程
result = slow_function() # 运行slow_function 行数，阻塞主线程。同时从属线程以动画形式旋转指针
# python 并没有提供终止线程的API，所以若想关闭线程，必须给线程发送消息。这里我们使用signal.go 属性：在主线程中把它设置为False后，spinner 线程会接收到，然后退出
signal.go = False
spinner.join() # 等待spinner 线程结束
return result
def main():
result = supervisor()
print('Answer', result)
if __name__ == '__main__':
main()
```
**协程**
```python
# spinner_asyncio.py
# 通过协程以动画的形式显示文本式旋转指针
import asyncio
import itertools
import sys
async def spin(msg):
write, flush = sys.stdout.write, sys.stdout.flush
for char in itertools.cycle('|/-\\'): # itertools.cycle 函数从指定的序列中反复不断地生成元素
status = char + ' ' + msg
write(status)
flush()
write('\x08' * len(status)) # 使用退格符把光标移回行首
try:
# 使用 yield from asyncio.sleep(0.1) 代替 time.sleep(.1), 这样的休眠不会阻塞事件循环
# 除非想阻塞主线程，从而冻结事件循环或整个应用，否则不要再 asyncio 协程中使用 time.sleep().如果协程需要在一段时间内什么都不做，应该使用 yield from asyncio.sleep(DELAY)
# 此处相当于另一协程
await asyncio.sleep(0.1)
# 如果 spin 函数苏醒后抛出 asyncio.CancelledError 异常，其原因是发出了取消请求
except asyncio.CancelledError as e:
print(str(e))
break
write(' ' * len(status) + '\x08' * len(status)) # 使用空格清除状态消息，把光标移回开头
async def slow_function(): # 5 现在此函数是协程，使用休眠假装进行I/O 操作时，使用 yield from 继续执行事件循环
# 假装等待I/O一段时间
await asyncio.sleep(3) # 此表达式把控制权交给主循环，在休眠结束后回复这个协程
return 42
# ?? 不能改为asynic supervisor 否则asyncio.async会报错 ，已找到原因已被asyncio.ensure_future替代??
async def supervisor():
spinner = asyncio.ensure_future(spin('thinking!')) # asyncio.async() 函数排定协程的运行时间，使用一个 Task 对象包装spin 协程，并立即返回
print('spinner object:', spinner) # Task 对象，输出类似 spinner object: <Task pending coro=<spin() running at spinner_asyncio.py:6>>
# 驱动slow_function() 函数，结束后，获取返回值。同事事件循环继续运行，
# 因为slow_function 函数最后使用yield from asyncio.sleep(3) 表达式把控制权交给主循环
result = await slow_function()
# Task 对象可以取消；取消后会在协程当前暂停的yield处抛出 asyncio.CancelledError 异常
# 协程可以捕获这个异常，也可以延迟取消，甚至拒绝取消
spinner.cancel()
return result
def main():
loop = asyncio.get_event_loop() # 获取事件循环引用
# 驱动supervisor 协程，让它运行完毕；这个协程的返回值是这次调用的返回值
result = loop.run_until_complete(supervisor())
loop.close()
print('Answer', result)
if __name__ == '__main__':
main()
```
**分析两段代码**
1. Task对象不由自己动手实例化，而是通过把协程传给 asyncio.async(...) 函数或 loop.create_task(...) 方法获取Task 对象已经排定了运行时间；而Thread 实例必须调用start方法，明确告知它运行
2. 在线程版supervisor函数中，slow_function 是普通的函数，由线程直接调用，而异步版的slow_function 函数是协程，由yield from 驱动。
3. 没有API能从外部终止线程，因为线程随时可能被中断。而协程如果想终止任务，可以使用Task.cancel() 实例方法，在协程内部抛出CancelledError 异常。协程可以在暂停的yield 处捕获这个异常，处理终止请求
4. supervisor 协程必须在main 函数中由loop.run_until_complete 方法执行。
5. **协程和线程相比关键的一个优点是**线程必须记住保留锁，去保护程序中的重要部分，防止多步操作在执行的过程中中断，而协程默认会做好保护，我们必须显式产出（使用yield 或 yield from 交出控制权）才能让程序的余下部分运行。
#### asyncio.Future：故意不阻塞
asyncio.Future 类与 concurrent.futures.Future 类的接口基本一致，不过实现方式不同，不可互换。在 concurrent.futures.Future 中，future只是调度执行某物的结果。在 asyncio 包中，BaseEventLoop.create_task(...) 方法接收一个协程，排定它的运行时间，然后返回一个asyncio.Task 实例（也是asyncio.Future 类的实例，因为 Task 是 Future 的子类，用于包装协程。（在 concurrent.futures.Future 中，类似的操作是Executor.submit(...)）。
与concurrent.futures.Future 类似，asyncio.Future 类也提供了：
* `.done()` 返回布尔值，表示Future 是否已经执行
* `.add_done_callback()` 这个方法只有一个参数，类型是可调用对象，Future运行结束后会回调这个对象。
* `.result()` 这个方法没有参数，因此不能指定超时时间。 如果调用 .result() 方法时期还没有运行完毕，会抛出 asyncio.InvalidStateError 异常。
### 协程嵌套 （协程的常用方式）
```python
import asyncio
import time
now = lambda: time.time()
async def do_some_work(x):
print('Waiting: ', x)
await asyncio.sleep(x)
return 'Done after {}s'.format(x)
async def main():
coroutine1 = do_some_work(1)
coroutine2 = do_some_work(2)
coroutine3 = do_some_work(4)
tasks = [
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3)
]
'''
# wait 有timeout参数
dones, pendings = await asyncio.wait(tasks)
for task in dones:
print('Task ret: ', task.result())
'''
# 如果使用的是 asyncio.gather创建协程对象，那么await的返回值就是协程运行的结果。
# 使用asyncio.wait(tasks)返回的顺序有点难以理解，但使用asyncio.gather(*tasks）返回值的顺序就好理解得多
results = await asyncio.gather(*tasks)
for result in results:
print('Task ret: ', result)
start = now()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
print('TIME: ', now() - start)
```
**抛出返回值到run_until_complete：**
```python
async def main():
coroutine1 = do_some_work(1)
coroutine2 = do_some_work(2)
coroutine3 = do_some_work(4)
tasks = [
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3)
]
return await asyncio.gather(*tasks)
start = now()
loop = asyncio.get_event_loop()
# 也可以直接返回到run_until_complete处理协程结果
results = loop.run_until_complete(main())
for result in results:
print('Task ret: ', result)
print('TIME: ', now() - start)
```
***使用as_completed**
```python
async def main(num):
tasks = []
i = 1
while i <= num :
tasks.append(asyncio.ensure_future(do_some_work(i)))
i += 1
for task in asyncio.as_completed(tasks):
result = await task
print('Task ret: {}'.format(result))
start = now()
loop = asyncio.get_event_loop()
done = loop.run_until_complete(main(10))
print('TIME: ', now() - start)
```
### 协程停止
future对象有几个状态：Pending,Running,Done,Cancelled. 创建future的时候，task为pending，事件循环调用执行的时候当然就是running，调用完毕自然就是done，如果需要停止事件循环，就需要先把task取消。可以使用asyncio.Task获取事件循环的task
```python
try:
loop = asyncio.get_event_loop()
loop.run_until_complete(main(5))
except KeyboardInterrupt as e:
for task in asyncio.Task.all_tasks():
# 启动事件循环之后，马上ctrl+c，会触发run_until_complete的执行异常 KeyBorardInterrupt。然后通过循环asyncio.Task取消future。
print(task)
# 返回true或false，已执行的返回false
print(task.cancel())
#loop stop之后还需要再次开启事件循环，最后再close，不然还会抛出异常：
# 抛出异常后要重新启动循环
loop.stop()
loop.run_forever()
finally:
loop.close()
```
**批量停止**
```python
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(main(5))
try:
loop.run_until_complete(task)
except KeyboardInterrupt as e:
print(asyncio.Task.all_tasks())
print('-------------------')
# 批量停止，如果全部停止成功就直接返回true，与上列不同
print(asyncio.gather(*asyncio.Task.all_tasks()).cancel())
loop.stop()
loop.run_forever()
finally:
loop.close()
```
### 不同线程的事件循环
很多时候，我们的事件循环用于注册协程，而有的协程需要动态的添加到事件循环中。一个简单的方式就是使用多线程。当前线程创建一个事件循环，然后在新建一个线程，在新线程中启动事件循环。当前线程不会被block。
```python
import asyncio
from threading import Thread
import time
now = lambda: time.time()
def start_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
def more_work(x):
print('More work {}'.format(x))
time.sleep(x)
print('Finished more work {}'.format(x))
start = now()
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
new_loop.call_soon_threadsafe(more_work, 6)
new_loop.call_soon_threadsafe(more_work, 3)
# 这里的计时没有意义，因为more_work具体的执行是在新的两个线程里面
print('TIME: {}'.format(now() - start))
```
启动上述代码之后，当前线程不会被block，新线程中会按照顺序执行call_soon_threadsafe方法注册的more_work方法，后者因为time.sleep操作是同步阻塞的，因此运行完毕more_work需要大致6 + 3
### 新线程协程
```python
def start_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
async def do_some_work(x):
print('Waiting {}'.format(x))
await asyncio.sleep(x)
print('Done after {}s'.format(x))
start = now()
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
asyncio.run_coroutine_threadsafe(do_some_work(6), new_loop)
asyncio.run_coroutine_threadsafe(do_some_work(4), new_loop)
print('TIME: {}'.format(time.time() - start))
```
上述的例子，主线程中创建一个new_loop，然后在另外的两个子线程中开启一个无限事件循环。主线程通过run_coroutine_threadsafe新注册协程对象。这样就能在子线程中进行事件循环的并发操作，同时主线程又不会被block。一共执行的时间大概在6s左右。
### master-worker主从模式
对于并发任务，通常是用生成消费模型，对队列的处理可以使用类似master-worker的方式，master主要用户获取队列的msg，worker用户处理消息。
为了简单起见，并且协程更适合单线程的方式，我们的主线程用来监听队列，子线程用于处理队列。这里使用redis的队列。主线程中有一个是无限循环，用户消费队列。
```python
while True:
task = rcon.rpop("queue")
if not task:
time.sleep(1)
continue
asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)
```
### 停止子线程
如果一切正常，那么上面的例子很完美。可是，需要停止程序，直接ctrl+c，会抛出KeyboardInterrupt错误，我们修改一下主循环：
```python
try:
while True:
task = rcon.rpop("queue")
if not task:
time.sleep(1)
continue
asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)
except KeyboardInterrupt as e:
print(e)
new_loop.stop()
```
可是实际上并不好使，虽然主线程try了KeyboardInterrupt异常，但是子线程并没有退出，为了解决这个问题，可以设置子线程为守护线程，这样当主线程结束的时候，子线程也随机退出。
```python
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.setDaemon(True) # 设置子线程为守护线程
t.start()
try:
while True:
# print('start rpop')
task = rcon.rpop("queue")
if not task:
time.sleep(1)
continue
asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)
except KeyboardInterrupt as e:
print(e)
new_loop.stop()
```
线程停止程序的时候，主线程退出后，子线程也随机退出才了，并且停止了子线程的协程任务。
```python
try:
while True:
# 用brpop方法，会block住task，如果主线程有消息，才会消费。
# 这种方式更适合队列消费，不用上面的要停顿一秒
_, task = rcon.brpop("queue")
asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)
except Exception as e:
print('error', e)
new_loop.stop()
finally:
pass
```
### 协程消费
主线程用于监听队列，然后子线程的做事件循环的worker是一种方式。还有一种方式实现这种类似master-worker的方案。即把监听队列的无限循环挪进协程中。程序初始化就创建若干个协程，实现类似并行的效果。一般这个方案就可以了
```python
import time
import asyncio
import redis
now = lambda : time.time()
# 最多开多少个协程
MAX_COROUTINES = 10
def get_redis():
connection_pool = redis.ConnectionPool(host='127.0.0.1', db=1,port=6379)
# connection_pool = redis.ConnectionPool(host='172.28.3.24', db=1,port=6379)
return redis.Redis(connection_pool=connection_pool)
rcon = get_redis()
async def worker():
print('Start worker')
while True:
start = now()
task = rcon.rpop("queue")
if not task:
await asyncio.sleep(1)
continue
print('Wait ', int(task))
await asyncio.sleep(int(task))
print('Done ', task, now() - start)
def main():
i = 0
while i < MAX_COROUTINES:
asyncio.ensure_future(worker())
i += 1
loop = asyncio.get_event_loop()
try:
loop.run_forever()
except KeyboardInterrupt as e:
print(asyncio.gather(*asyncio.Task.all_tasks()).cancel())
loop.stop()
loop.run_forever()
finally:
loop.close()
if __name__ == '__main__':
main()
```
## aiohttp
asyncio可以实现单线程并发IO操作。如果仅用在客户端，发挥的威力不大。如果把asyncio用在服务器端，例如Web服务器，由于HTTP连接就是IO操作，因此可以用单线程+coroutine实现多用户的高并发支持。
asyncio实现了TCP、UDP、SSL等协议，aiohttp则是基于asyncio实现的HTTP框架。
### 实现web服务器
```python
import asyncio
from aiohttp import web
async def index(request):
await asyncio.sleep(0.5)
return web.Response(body=b'<h1>Index</h1>')
async def hello(request):
await asyncio.sleep(0.5)
text = '<h1>hello, %s!</h1>' % request.match_info['name']
return web.Response(body=text.encode('utf-8'))
async def init(loop):
app = web.Application(loop=loop)
app.router.add_route('GET', '/', index)
app.router.add_route('GET', '/hello/{name}', hello)
srv = await loop.create_server(app.make_handler(), '127.0.0.1', 8000)
print('Server started at http://127.0.0.1:8000...')
return srv
loop = asyncio.get_event_loop()
loop.run_until_complete(init(loop))
loop.run_forever()
```
## 例1，[使用asyncio 和 aiohttp 包下载国旗][1]
## 例2，[使用python-aiohttp爬取网易云音乐，今日头条，搭建微信公众平台][2]
[1]: https://gitee.com/nixi8/Python/tree/master/script
[2]: https://github.com/SigalHu/WeiXin
