理解生成器
定义生成器
yield关键字，可以让我们定义一个生成器函数。
def generator_func(): print('a') yield 1 g = generator_func() print(g)
>>> <generator object generator_func at 0x10e178b88>
推动生成器
使用next函数从生成器中取值
def generator_func(): print('a') yield 1 g = generator_func() ret1 = next(g) print(ret1)
>>>
a
1
使用next可以推动生成器的执行，下面的代码，我们可以看到每一次执行next可以让generator_func中的代码从上一个位置开始继续执行到yield，并且将yield后面的值返回到函数外部，最终我们可以执行到yield 3。
def generator_func(): print('a') yield 1 print('b') yield 2 print('c') yield 3 print('d') g = generator_func() ret1 = next(g) print(ret1) ret2 = next(g) print(ret2) ret3 = next(g) print(ret3)
>>>
a
1
b
2
c
3
当函数中已经没有更多的yield时继续执行next(g)，遇到StopIteration
def generator_func(): print('a') yield 1 print('b') yield 2 print('c') yield 3 print('d') g = generator_func() ret1 = next(g) print(ret1) ret2 = next(g) print(ret2) ret3 = next(g) print(ret3) next(g)
send向生成器中发送数据。send的作用相当于next，只是在驱动生成器继续执行的同时还可以向生成器中传递数据。
import numbers def cal_sum(): sum_num = 0 while True: num = yield if isinstance(num,numbers.Integral): sum_num += num print('sum :',sum_num) elif num is None: break g = cal_sum() g.send(None) # 相当于next(g),预激活生成器 g.send(31) g.send(25) g.send(17) g.send(8) >>> sum : 31 sum : 56 sum : 73 sum : 81
生成器中的return和StopIteration
import numbers def cal_sum(): sum_num = 0 while True: num = yield if isinstance(num,numbers.Integral): sum_num += num print('sum :',sum_num) elif num is None: break return sum_num g = cal_sum() g.send(None) # 相当于next(g),预激活生成器 g.send(31) g.send(25) g.send(17) g.send(8) g.send(None) # 停止生成器 >>> sum : 31 sum : 56 sum : 73 sum : 81 Traceback (most recent call last): File "/Users/jingliyang/PycharmProjects/python的进阶/manager.py", line 19, in <module> g.send(None) StopIteration: 81
import numbers def cal_sum(): sum_num = 0 while True: num = yield if isinstance(num,numbers.Integral): sum_num += num print('sum :',sum_num) elif num is None: break return sum_num g = cal_sum() g.send(None) # 相当于next(g),预激活生成器 g.send(31) g.send(25) g.send(17) g.send(8) try: g.send(None) # 停止生成器 except StopIteration as e: print(e.value)
生成器的close和throw
使用throw向生成器中抛一个异常
def throw_test(): print('a') yield 1 print('b') yield 2 g = throw_test() next(g) g.throw(Exception,'value error')
>>>
a
Traceback (most recent call last):
File "/Users/jingliyang/PycharmProjects/python的进阶/manager.py", line 32, in <module>
g.throw(ValueError,'value error') # throw和send、next相同，都是驱动生成器继续执行，只不过throw用来向生成器中抛一个异常
File "/Users/jingliyang/PycharmProjects/python的进阶/manager.py", line 26, in throw_test
yield 1
ValueError: value error
def throw_test(): print('a') try: yield 1 except ValueError: pass print('b') yield 2 g = throw_test() next(g) ret = g.throw(ValueError,'value error') # throw和send、next相同，都是驱动生成器继续执行，只不过throw用来向生成器中抛一个异常 print(ret) >>> a b 2
使用close关闭一个生成器
def throw_test(): print('a') yield 1 print('b') yield 2 g = throw_test() ret1 = next(g) print(ret1) g.close() next(g) >>> a 1 Traceback (most recent call last): File "/Users/jingliyang/PycharmProjects/python的进阶/manager.py", line 45, in <module> next(g) StopIteration
yield from关键字
yield from关键字可以直接返回一个生成器
l = ['h','e','l'] dic = {'l':'v1','o':'v2'} s = 'eva' def yield_from_gen(): for i in l: yield i for j in dic: yield j for k in s: yield k for item in yield_from_gen(): print(item,end='') >>>helloeva l = ['h','e','l'] dic = {'l':'v1','o':'v2'} s = 'eva' def yield_from_gen(): yield from l yield from dic yield from s for item in yield_from_gen(): print(item,end='') >>>helloeva
from itertools import chain l = ['h','e','l'] dic = {'l':'v1','o':'v2'} s = 'eva' def yield_from_gen(): yield from chain(l,dic,s) for item in yield_from_gen(): print(item,end='')
利用yield from完成股票的计算，yield from能够完成一个委派生成器的作用，在子生成器和调用者之间建立起一个双向通道。
def son_gen(): avg_num = 0 sum_num = 0 count = 1 while True: num = yield avg_num if num: sum_num += num avg_num = sum_num/count count += 1 else:break return avg_num def depute_gen(key): while True: ret = yield from son_gen() print(key,ret) def main(): shares_list= { 'sogou':[6.4,6.5,6.6,6.2,6.1,6.6,6.7], 'alibaba':[181.72,184.58,183.54,180,88,169.88,178.21,189.32], '美团':[59.7,52.6,47.2,55.4,60.7,66.1,74.0] } for key in shares_list: g = depute_gen(key) next(g) for v in shares_list[key]: g.send(v) g.send(None) main()
协程
概念
根据维基百科给出的定义，“协程 是为非抢占式多任务产生子程序的计算机程序组件，协程允许不同入口点在不同位置暂停或开始执行程序”。从技术的角度来说，“协程就是你可以暂停执行的函数”。如果你把它理解成“就像生成器一样”，那么你就想对了。
使用yield实现协程
#基于yield实现异步 import time def consumer(): '''任务1:接收数据,处理数据''' while True: x=yield def producer(): '''任务2:生产数据''' g=consumer() next(g) for i in range(10000000): g.send(i) producer()
使用yield from实现的协程
import datetime import heapq # 堆模块 import types import time class Task: def __init__(self, wait_until, coro): self.coro = coro self.waiting_until = wait_until def __eq__(self, other): return self.waiting_until == other.waiting_until def __lt__(self, other): return self.waiting_until < other.waiting_until class SleepingLoop: def __init__(self, *coros): self._new = coros self._waiting = [] def run_until_complete(self): for coro in self._new: wait_for = coro.send(None) heapq.heappush(self._waiting, Task(wait_for, coro)) while self._waiting: now = datetime.datetime.now() task = heapq.heappop(self._waiting) if now < task.waiting_until: delta = task.waiting_until - now time.sleep(delta.total_seconds()) now = datetime.datetime.now() try: print('*'*50) wait_until = task.coro.send(now) print('-'*50) heapq.heappush(self._waiting, Task(wait_until, task.coro)) except StopIteration: pass
def sleep(seconds): now = datetime.datetime.now() wait_until = now + datetime.timedelta(seconds=seconds) print('before yield wait_until') actual = yield wait_until # 返回一个datetime数据类型的时间 print('after yield wait_until') return actual - now def countdown(label, length, *, delay=0): print(label, 'waiting', delay, 'seconds before starting countdown') delta = yield from sleep(delay) print(label, 'starting after waiting', delta) while length: print(label, 'T-minus', length) waited = yield from sleep(1) length -= 1 print(label, 'lift-off!') def main(): loop = SleepingLoop(countdown('A', 5), countdown('B', 3, delay=2), countdown('C', 4, delay=1)) start = datetime.datetime.now() loop.run_until_complete() print('Total elapsed time is', datetime.datetime.now() - start) if __name__ == '__main__': main()
await和async关键字
使用 async function 可以定义一个 异步函数，在async关键字定义的函数中不能出现yield和yield from
# 例1 async def download(url): # 加入新的关键字 async ，可以将任何一个普通函数变成协程 return 'eva' ret = download('http://www.baidu.com/') print(ret) # <coroutine object download at 0x108b3a5c8> ret.send(None) # StopIteration: eva # 例2 async def download(url): return 'eva' def run(coroutine): try: coroutine.send(None) except StopIteration as e: return e.value coro = download('http://www.baidu.com/') ret = run(coro) print(ret)
async关键字不能和yield一起使用，引入coroutine装饰器来装饰downloader生成器。
await 操作符后面必须跟一个awaitable对象（通常用于等待一个会有io操作的任务）, 它只能在异步函数 async function 内部使用。
# 例3 import types @types.coroutine # 将一个生成器变成一个awaitable的对象 def downloader(url): yield 'aaa' async def download_url(url): # 协程 waitable = downloader(url) print(waitable) # <generator object downloader at 0x1091e2c78> 生成器 html = await waitable return html coro = download_url('http://www.baidu.com') print(coro) # <coroutine object download_url at 0x1091c9d48> ret = coro.send(None) print(ret)
asyncio模块
asyncio是Python 3.4版本引入的标准库，直接内置了对异步IO的支持。
asyncio的编程模型就是一个消息循环。我们从asyncio模块中直接获取一个EventLoop的引用，然后把需要执行的协程扔到EventLoop中执行，就实现了异步IO。
coroutine+yield from
import asyncio @asyncio.coroutine def hello(): print("Hello world!") # 异步调用asyncio.sleep(1): r = yield from asyncio.sleep(1) print("Hello again!") # 获取EventLoop: loop = asyncio.get_event_loop() # 执行coroutine loop.run_until_complete(hello()) loop.close()
async+await
import asyncio async def hello(): print("Hello world!") # 异步调用asyncio.sleep(1): r = await asyncio.sleep(1) print("Hello again!") # 获取EventLoop: loop = asyncio.get_event_loop() # 执行coroutine loop.run_until_complete(hello()) loop.close()
执行多个任务
import asyncio async def hello(): print("Hello world!") await asyncio.sleep(1) print("Hello again!") return 'done' loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait([hello(),hello()])) loop.close()
获取返回值
import asyncio async def hello(): print("Hello world!") await asyncio.sleep(1) print("Hello again!") return 'done' loop = asyncio.get_event_loop() task = loop.create_task(hello()) loop.run_until_complete(task) ret = task.result() print(ret)
执行多个任务获取返回值
import asyncio async def hello(i): print("Hello world!") await asyncio.sleep(i) print("Hello again!") return 'done',i loop = asyncio.get_event_loop() task1 = loop.create_task(hello(2)) task2 = loop.create_task(hello(1)) task_l = [task1,task2] tasks = asyncio.wait(task_l) loop.run_until_complete(tasks) for t in task_l: print(t.result())
执行多个任务按照返回的顺序获取返回值
import asyncio async def hello(i): print("Hello world!") await asyncio.sleep(i) print("Hello again!") return 'done',i async def main(): tasks = [] for i in range(20): tasks.append(asyncio.ensure_future(hello((20-i)/10))) for res in asyncio.as_completed(tasks): result = await res print(result) loop = asyncio.get_event_loop() loop.run_until_complete(main())
asyncio使用协程完成http访问
import asyncio async def get_url(): reader,writer = await asyncio.open_connection('www.baidu.com',80) writer.write(b'GET / HTTP/1.1 HOST:www.baidu.com Connection:close ') all_lines = [] async for line in reader: data = line.decode() all_lines.append(data) html = ' '.join(all_lines) return html async def main(): tasks = [] for url in range(20): tasks.append(asyncio.ensure_future(get_url())) for res in asyncio.as_completed(tasks): result = await res print(result) if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(main()) # 处理一个任务 loop.run_until_complete(asyncio.wait([main()])) # 处理多个任务 task = loop.create_task(main()) # 使用create_task获取返回值 loop.run_until_complete(task) loop.run_until_complete(asyncio.wait([task]))