深究Python中的asyncio库 Asyncio ——gather vs wait在Asyncio中不止可以多次使用asyncio.gather，还有另外一个用法是asyncio.wait，他们都可以让多个协程并发执行。那为什么提供2个方法呢？他们有什么区别，适用场景是怎么样的呢？我们先看2个协程的例子:async def a():     print('Suspending a')     await asyncio.sleep(3)     print('Resuming a')     return 'A' async def b():     print('Suspending b')     await asyncio.sleep(1)     print('Resuming b')     return 'B'在IPython里面用gather执行一下：In : return_value_a, return_value_b = await asyncio.gather(a(), b()) Suspending a Suspending b Resuming b Resuming a In : return_value_a, return_value_b Out: ('A', 'B')Ok，asyncio.gather方法的名字说明了它的用途，gather的意思是「搜集」，也就是能够收集协程的结果，而且要注意，它会按输入协程的顺序保存的对应协程的执行结果。接着我们说asyncio.await，先执行一下：In : done, pending = await asyncio.wait([a(), b()]) Suspending b Suspending a Resuming b Resuming a In : done Out: {<Task finished coro=<a() done, defined at <ipython-input-5-5ee142734d16>:1> result='A'>,  <Task finished coro=<b() done, defined at <ipython-input-5-5ee142734d16>:8> result='B'>} In : pending Out: set() In : task = list(done)[0] In : task Out: <Task finished coro=<b() done, defined at <ipython-input-5-5ee142734d16>:8> result='B'> In : task.result() Out: 'B'asyncio.wait的返回值有2项，第一项表示完成的任务列表(done)，第二项表示等待(Future)完成的任务列表(pending)，每个任务都是一个Task实例，由于这2个任务都已经完成，所以可以执行task.result()获得协程返回值。Ok, 说到这里，总结下它俩的区别的第一层区别:asyncio.gather封装的Task全程黑盒，只告诉你协程结果。asyncio.wait会返回封装的Task(包含已完成和挂起的任务)，如果你关注协程执行结果你需要从对应Task实例里面用result方法自己拿。为什么说「第一层区别」，asyncio.wait看名字可以理解为「等待」，所以返回值的第二项是pending列表，但是看上面的例子，pending是空集合，那么在什么情况下，pending里面不为空呢？这就是第二层区别：asyncio.wait支持选择返回的时机。asyncio.wait支持一个接收参数return_when，在默认情况下，asyncio.wait会等待全部任务完成(return_when='ALL_COMPLETED')，它还支持FIRST_COMPLETED（第一个协程完成就返回）和FIRST_EXCEPTION（出现第一个异常就返回）：In : done, pending = await asyncio.wait([a(), b()], return_when=asyncio.tasks.FIRST_COMPLETED) Suspending a Suspending b Resuming b In : done Out: {<Task finished coro=<b() done, defined at <ipython-input-5-5ee142734d16>:8> result='B'>} In : pending Out: {<Task pending coro=<a() running at <ipython-input-5-5ee142734d16>:3> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x108065e58>()]>>}看到了吧，这次只有协程b完成了，协程a还是pending状态。在大部分情况下，用asyncio.gather是足够的，如果你有特殊需求，可以选择asyncio.wait，举2个例子：需要拿到封装好的Task，以便取消或者添加成功回调等业务上需要FIRST_COMPLETED/FIRST_EXCEPTION即返回的asyncio.create_task vs loop.create_task vs asyncio.ensure_future创建一个Task一共有3种方法，如这小节的标题。在上篇文章我说过，从Python 3.7开始可以统一的使用更高阶的asyncio.create_task。其实asyncio.create_task就是用的loop.create_task：def create_task(coro):     loop = events.get_running_loop()     return loop.create_task(coro)loop.create_task接受的参数需要是一个协程，但是asyncio.ensure_future除了接受协程，还可以是Future对象或者awaitable对象:如果参数是协程，其实底层还是用的loop.create_task，返回Task对象如果是Future对象会直接返回如果是一个awaitable对象会await这个对象的__await__方法，再执行一次ensure_future，最后返回Task或者Future所以就像ensure_future名字说的，确保这个是一个Future对象：Task是Future 子类，前面说过一般情况下开发者不需要自己创建Future其实前面说的asyncio.wait和asyncio.gather里面都用了asyncio.ensure_future。对于绝大多数场景要并发执行的是协程，所以直接用asyncio.create_task就足够了~下一节：深究Python中的asyncio库-shield函数

深究Python中的asyncio库 shieldasyncio.shield，用它可以屏蔽取消操作。一直到这里，我们还没有见识过Task的取消。看一个例子:In : loop = asyncio.get_event_loop() In : task1 = loop.create_task(a()) In : task2 = loop.create_task(b()) In : task1.cancel() Out: True In : await asyncio.gather(task1, task2) Suspending a Suspending b --------------------------------------------------------------------------- CancelledError                            Traceback (most recent call last) cell_name in async-def-wrapper() CancelledError:在上面的例子中，task1被取消了后再用asyncio.gather收集结果，直接抛CancelledError错误了。这里有个细节，gather支持return_exceptions参数：In : await asyncio.gather(task1, task2, return_exceptions=True) Out: [concurrent.futures._base.CancelledError(), 'B']可以看到，task2依然会执行完成，但是task1的返回值是一个CancelledError错误，也就是任务被取消了。如果一个创建后就不希望被任何情况取消，可以使用asyncio.shield保护任务能顺利完成。不过要注意一个陷阱，先看错误的写法：In : task1 = asyncio.shield(a()) In : task2 = loop.create_task(b()) In : task1.cancel() Out: True In : await asyncio.gather(task1, task2, return_exceptions=True) Suspending a Suspending b Resuming b Out: [concurrent.futures._base.CancelledError(), 'B']可以看到依然是CancelledError错误，且协程a未执行完成，正确的用法是这样的：In : task1 = asyncio.shield(a()) In : task2 = loop.create_task(b()) In : ts = asyncio.gather(task1, task2, return_exceptions=True) In : task1.cancel() Out: True In : await ts Suspending a Suspending b Resuming a Resuming b Out: [concurrent.futures._base.CancelledError(), 'B']可以看到虽然结果是一个CancelledError错误，但是看输出能确认协程实际上是执行了的。所以正确步骤是：先创建 GatheringFuture 对象 ts取消任务await tsasynccontextmanager如果你了解Python，之前可能听过或者用过contextmanager ，一个上下文管理器。通过一个计时的例子就理解它的作用:from contextlib import contextmanager async def a():     await asyncio.sleep(3)     return 'A' async def b():     await asyncio.sleep(1)     return 'B' async def s1():     return await asyncio.gather(a(), b()) @contextmanager def timed(func):     start = time.perf_counter()     yield asyncio.run(func())     print(f'Cost: {time.perf_counter() - start}')timed函数用了contextmanager装饰器，把协程的运行结果yield出来，执行结束后还计算了耗时：In : from contextmanager import * In : with timed(s1) as rv: ...:     print(f'Result: {rv}') ...: Result: ['A', 'B'] Cost: 3.0052654459999992大家先体会一下。在Python 3.7添加了asynccontextmanager，也就是异步版本的contextmanager，适合异步函数的执行，上例可以这么改：@asynccontextmanager async def async_timed(func):     start = time.perf_counter()     yield await func()     print(f'Cost: {time.perf_counter() - start}') async def main():     async with async_timed(s1) as rv:         print(f'Result: {rv}') In : asyncio.run(main()) Result: ['A', 'B'] Cost: 3.00414147500004async版本的with要用async with，另外要注意yield await func()这句，相当于yield + await func()PS: contextmanager 和 asynccontextmanager 最好的理解方法是去看源码注释下一节：深究Python中的asyncio库-函数的回调与调度

深究Python中的asyncio库 前面的代码都是异步的，就如sleep，需要用asyncio.sleep而不是阻塞的time.sleep，如果有同步逻辑，怎么利用asyncio实现并发呢？答案是用run_in_executor。在一开始我说过开发者创建 Future 对象情况很少，主要是用run_in_executor，就是让同步函数在一个执行器( executor)里面运行。同步代码def a():     time.sleep(1)     return 'A' async def b():     await asyncio.sleep(1)     return 'B' def show_perf(func):     print('*' * 20)     start = time.perf_counter()     asyncio.run(func())     print(f'{func.__name__} Cost: {time.perf_counter() - start}') async def c1():     loop = asyncio.get_running_loop()     await asyncio.gather(         loop.run_in_executor(None, a),         b()     ) In : show_perf(c1) ******************** c1 Cost: 1.0027242230000866可以看到用run_into_executor可以把同步函数逻辑转化成一个协程，且实现了并发。这里要注意细节，就是函数a是普通函数，不能写成协程，下面的定义是错误的，不能实现并发：async def a():     time.sleep(1)     return 'A'因为 a 里面没有异步代码，就不要用async def来定义。需要把这种逻辑用loop.run_in_executor封装到协程：async def c():     loop = asyncio.get_running_loop()     return await loop.run_in_executor(None, a)大家理解了吧？loop.run_in_executor(None, a)这里面第一个参数是要传递concurrent.futures.Executor实例的，传递None会选择默认的executor：In : loop._default_executor Out: <concurrent.futures.thread.ThreadPoolExecutor at 0x112b60e80>当然我们还可以用进程池，这次换个常用的文件读写例子，并且用:async def c3():     loop = asyncio.get_running_loop()     with concurrent.futures.ProcessPoolExecutor() as e:         print(await asyncio.gather(             loop.run_in_executor(e, a),             b()         )) In : show_perf(c3) ******************** ['A', 'B'] c3 Cost: 1.0218078890000015下一节：深究Python中的asyncio库-线程池

深究Python中的asyncio库 在同步线程中使用的run_in_executor就如它方法的名字所示，把协程放到了一个执行器里面，可以在一个线程池，也可以在一个进程池。另外还可以使用run_coroutine_threadsafe在其他线程执行协程（这是线程安全的）。多线程def start_loop(loop):     asyncio.set_event_loop(loop)     loop.run_forever() def shutdown(loop):     loop.stop() async def b1():     new_loop = asyncio.new_event_loop()     t = Thread(target=start_loop, args=(new_loop,))     t.start()     future = asyncio.run_coroutine_threadsafe(a(), new_loop)     print(future)     print(f'Result: {future.result(timeout=2)}')     new_loop.call_soon_threadsafe(partial(shutdown, new_loop)) In : await b1() <Future at 0x107edf4e0 state=pending> Result: A这里面有几个细节要注意:协程应该从另一个线程中调用，而非事件循环运行所在线程，所以用asyncio.new_event_loop()新建一个事件循环在执行协程前要确保新创建的事件循环是运行着的，所以需要用start_loop之类的方式启动循环接着就可以用asyncio.run_coroutine_threadsafe执行协程a了，它返回了一个Future对象可以通过输出感受到future一开始是pending的，因为协程a里面会sleep 1秒才返回结果用future.result(timeout=2)就可以获得结果，设置timeout的值要大于a协程执行时间，要不然会抛出TimeoutError一开始我们创建的新的事件循环跑在一个线程里面，由于loop.run_forever会阻塞程序关闭，所以需要结束时杀掉线程，所以用call_soon_threadsafe回调函数shutdown去停止事件循环这里再说一下call_soon_threadsafe，看名字就知道它是线程安全版本的call_soon，其实就是在另外一个线程里面调度回调。BTW， 其实asyncio.run_coroutine_threadsafe底层也是用的它。

详解Python元类(metaclass) 什么是元类？理解元类（metaclass）之前，我们先了解下Python中的OOP和类（Class）。面向对象全称 Object Oriented Programming 简称OOP，这种编程思想被大家所熟知。它是把对象作为一个程序的基本单元，把数据和功能封装在里面，能够实现很好的复用性，灵活性和扩展性。OOP中有2个基本概念：类和对象：类是描述如何创建一个对象的代码段，用来描述具有相同的属性和方法的对象的集合，它定义了该集合中每个对象所共有的属性和方法对象是类的实例（Instance）。我们举个例子：In : class ObjectCreator(object): ...:     pass ...: In : my_object = ObjectCreator() In : my_object Out: <__main__.ObjectCreator at 0x1082bbef0>而Python中的类并不是仅限于此：In : print(ObjectCreator) <class '__main__.ObjectCreator'>ObjectCreator竟然可以被print，所以它的类也是对象！既然类是对象，你就能动态地创建它们，就像创建任何对象那样。我在日常工作里面就会有这种动态创建类的需求，比如在mock数据的时候，现在有个函数func接收一个参数：In : def func(instance): ...:     print(instance.a, instance.b) ...:     print(instance.method_a(10)) ...:正常使用起来传入的instance是符合需求的（有a、b属性和method_a方法），但是当我想单独调试func的时候，需要「造」一个，假如不用元类，应该是这样写:In : def generate_cls(a, b): ...:     class Fake(object): ...:         def method_a(self, n): ...:             return n ...:     Fake.a = a ...:     Fake.b = b ...:     return Fake ...: In : ins = generate_cls(1, 2)() In : ins.a, ins.b, ins.method_a(10) Out: (1, 2, 10)你会发现这不算算是「动态创建」的：类名（Fake）不方便改变要创建的类需要的属性和方法越多，就要对应的加码，不灵活。我平时怎么做呢：In : def method_a(self, n): ...:     return n ...:  In : ins = type('Fake', (), {'a': 1, 'b': 2, 'method_a': method_a})() In : ins.a, ins.b, ins.method_a(10) Out: (1, 2, 10)到了这里，引出了type函数。本来它用来能让你了解一个对象的类型：In : type(1) Out: int In : type('1') Out: str In : type(ObjectCreator) Out: type In : type(ObjectCreator()) Out: __main__.ObjectCreator另外，type如上所说还可以动态地创建类：type可以把对于类的描述作为参数，并返回一个类。用来创建类的东东就是「元类」MyClass = type('MyClass', (), {})这种用法就是由于type实际上是一个元类，作为元类的type在Python中被用于在后台创建所有的类。在Python语言上有个说法「Everything is an object」。包整数、字符串、函数和类... 所有这些都是对象。所有这些都是由一个类创建的：In : age = 35 In : age.__class__ Out: int In : name = 'bob' In : name.__class__ Out: str ...现在，任何__class__中的特定__class__是什么？In : age.__class__.__class__ Out: type In : name.__class__.__class__ Out: type ...如果你愿意，你可以把type称为「类工厂」。type是Python中内建元类，当然，你也可以创建你自己的元类。创建自己的元类Python2创建类的时候，可以添加一个__metaclass__属性：class Foo(object):     __metaclass__ = something...     [...]如果你这样做，Python会使用元类来创建Foo这个类。Python会在类定义中寻找__metaclass__。如果找到它，Python会用它来创建对象类Foo。如果没有找到它，Python将使用type来创建这个类。在Python3中语法改变了一下：class Simple1(object, metaclass=something...):     [...]本质上是一样的。拿一个4年前写分享的元类例子（就是为了推荐你来阅读