1 简介

本小节介绍异步编程常用的关键用法和python中的“方言”。

2 同步原语 Synchronization primitives

以下是一些关键同步原语。

锁：

    Lock
    Event
    Condition

信号量：

    Semaphore
    BoundedSemaphore
    ASYNCIO锁API被设计成接近的类threading 模块（Lock，Event， Condition，
    Semaphore， BoundedSemaphore），但它没有超时参数。该 asyncio.wait_for()功能可用于在超时后取消任务。

2.1 锁具

2.1.1。锁 Lock

示例

	class asyncio.Lock(*, loop=None)   # 非线程安全
		# 原始锁对象，基元锁是一种同步基元。
		# 原始锁只处于锁定 解锁两种状态 
		  # 协程 已获得 acquire(),  与yield from 一起使用
		  # release()
		yield from lock  # 锁支持上下文管理协议。应该用作上下文管理表达式

• 获得锁

    lock = Lock()
    ...
    yield from lock
    try:
    	...
    finally:
  

• 释放锁

        lock.release()
  

• 上下文管理器

    lock = Lock()
    ....
    with (yield from lock):
       ...
  

• 测试锁定对象的锁定状态

    if not lock.locked():
        yield from lock
    else:
       # lock is acquired
  

功能：

	是否获得锁 locked()   # 如果获得了锁，返回True
	协程 获取锁 acquire()
	释放锁 release()

2.1.2 事件 Event

Event 实现，异步等效于 threading.Event,实现事件对象的类。
事件管理一个标志(默认为false)，该标志可用通过方法设置为true，并通过set()设置为false

    class asyncio.Event(*, loop=None)   # 非线程安全

方法

        clear()   # 内部标志位重置false，协程调用wait()将阻塞到set()被调用以再次将内部标志设置位true
        is_set() #True当且仅当北部标志为true才返回
        set()  # 将内部标志设置为true
        wait()   # 协程

2.1.3。 触发条件 Condition

触发 条件的实现，异步等效于threading.Condition。

此类实现条件变量对象，条件变量运行一个或多个协程等待，直达他们被另一个协程通知。

如果没有给出lock参数为None，则必须是一个Lock对象，并且用作基础锁，否则将Lock创建一个新对象并将其用作基础锁

    class asyncio.Condition(lock=None, *, loop=None)  # 非线程安全

获取基础锁，此方法将阻塞到解锁，然后再将其设置为lock并返回True

        协程 acquire()    

默认情况夏，唤醒一个协程等待这个情况(如果有)如果在调用此方法时调用协程尚未获取锁定 引发错误 RuntimeError

        notify(n=1)  

如获得了基础锁，则返回

        locked()   

唤醒等待这种情况的所有协程。此方法类似 notify() 但此方法是唤醒所有等待的协程，如果调用此方法时被调用协程尚未获取锁定，则触发 RuntimeError

        notify_all()   

无返回，释放基础锁，将其重置为解锁状态然后返回。未锁定的锁被调用时，引发RuntimeError

        release()   

等到收到通知，如果调用此方法时协程尚未获取锁定，

        asyncio wait()   

此方法释放基础锁，然后进行阻塞 直到被notify()等类似方法唤醒为止。唤醒后，它将重新获取锁并返回True

        RuntimeError将触发

等到func变为真，func应为可调用的，结果解释为布尔值

        asyncio wait_for(func)   

2.2 信号量 Semaphores

2.2.1。信号

Semaphore 信号量实现。 此类管理一个内部计数器，计数器由每个acquire()调用递减，并由每个release()调用递增。

计数器永远不能低于0，当acquire()发现它为0时，将阻塞，直到其他协程调用release() 为止。

		class asyncio.Semaphore(value=1, *, loop=None)   
                    # 非线程安全

acquire 获取一个信号量，如果内部计数器在输入时大于零，将其递减一，并返回True。如果进入时为0，阻塞；等待其他协程调用release() 使其大于0，然后返回True

		asyncio acquire()    

locked 如果无法立即获取信号量返回 True

		locked()   

release释放信号量，使内部计数器加1，进入时为0，并且另一个协程正则等待再次变为0，唤醒该协程

		release()    

2.2.2。有界信号量

有界信号量实现，继承自Semaphore。 在release() 它是否将增加超过ValueError初始值的值

		class asyncio.BoundedSemaphore(value=1, *, loop=None)

2.3. 队列集

源代码： Lib / asyncio / queues.py

	Queue
	PriorityQueue
	LifeQueue

ASYNCIO队列API被设计为接近的queue模块的类 asyncio.wait_for() 功能可用于在超时 后取消任务

2.3.1. 队列

队列用于协程生产者和消费者协程，如果maxsize小于或等于零，队列大小无限。 如果maxsize大于0，当队列达到maxsize时将阻塞，直到被删除 yield from

	class asyncio.Queue(Maxsize=0, *, loop=None)   # >py3.44

压入和获取 put get
与标准库不同queue，可用可靠的知晓Queue的大小qsize()，因为单线程asyncio应用程序不会在 调用qsize()和在Queue上执行操作时被中断。

		put(), get()

empty 如果队列为空返回True，否则返回False

		empty()  

full 如果有maxsize个条目在队列，则返回True。 如果Queue使用参数maxsize=0， 则full()用于不会为True

		full()    

异步 get 从队列删除并返回一个元素。如果队列为空，则等待，直到队列有元素
async get()

get_nowait从队列中删除并返回一个项目，立即返回一个队列中的元素，如果队列有值，引发异常QueueEmpty

		 get_nowait()  

async join 阻塞直到队列所有项目都已获得并处理，每当将项目添加到队列时，未完成任务数量将增加。 当未完成任务数降至0，join() 取消阻止
async join()

async put 项目放入队列。如果队列已满，请等待空闲插槽可用再添加到项目

		async put()  

put_nowait 不阻塞的放一个元素入队列。如果没有空闲槽位，引发QueueFull异常

		put_nowait()  

qsize 队列中的项目数

		qsize()  

task_done 前面排队的任务已经完成，即get出来的元素相关操作已经完成。

		task_done() # >py3.4.4  

maxsize 队列中可存放的元素数量

		maxsize  

2.3.2。PriorityQueue

PriorityQueue子类Queue，以优先级顺序检索条目 从低到低，条目通常是以元组形式 （优先级，数据）。

	class asyncio.PriorityQueue

LifoQueue 子类Queue首先检索最近添加的条目

	class asyncio.LifoQueue

异常

exception asyncio.TimeoutError

        该操作已超过规定的截止日期。
        重要 这个异常与内置 TimeoutError 异常不同。

exception asyncio.CancelledError

       该操作已被取消。
        取消asyncio任务时，可以捕获此异常以执行自定义操作。
        在几乎所有情况下，都必须重新引发异常。
        在 3.8 版更改: CancelledError 现在是 BaseException 的子类。

exception asyncio.InvalidStateError

        Task 或 Future 的内部状态无效。
        在为已设置结果值的未来对象设置结果值等情况下，可以引发此问题。

exception asyncio.SendfileNotAvailableError

        "sendfile" 系统调用不适用于给定的套接字或文件类型。
        子类 RuntimeError 。

exception asyncio.IncompleteReadError

        请求的读取操作未完全完成。
        由 asyncio stream APIs 提出
        此异常是 EOFError 的子类。	

expected

        预期字节的总数（ int ）。

partial

        到达流结束之前读取的 bytes 字符串。

exception asyncio.LimitOverrunError

        在查找分隔符时达到缓冲区大小限制。
        由 asyncio stream APIs 提出
        consumed
                要消耗的字节总数。

asyncio.QueueEmpty

		get_nowait() 时，Queue为空对象时引发

asyncio.QueueFull

		put_nowait() 在Queue已满的对象上调用该方法时引发异常

3 使用 用asyncio开发

异步与传统的顺序 编程不同. 这里有一些常见的错误和陷阱,如何避免它们.

 异步调试模式
 消除
 并发和多线程
 正确处理阻塞功能
 日志记录
 检测从未调度的协程对象
 检测从未消耗的异常
 正确协程
 待处理任务已销毁
 关闭传输和事件循环

3.1 Debug模式

默认场景下,asyncio 以生成模式运行,为了简化开发,asyncio还有一种debug模式

    将 PYTHONASYNCIODEBUG 设置为 1
    使用 python 开发模式
    将debug=True 传递给 asyncio.run()
    调用 loop.set_debug()

调试模式将有以下效应

asyncio 检查 未被等待的协程 并记录他们,这将消除被遗忘的等待的问题.
许多非线程安全的异步APIs ，例如loop.call_soon(), loop._call_at(), 如果从错误线程调用,则引发异常.

如果执行I/O操作花费时间太长,则记录I/O选择器的执行时间.
执行时间超过100毫秒的回调将载入日志.

属性 loop.slow_callback_duration 可用于设置秒为单位的最小执行持续时间. 这表示 缓慢.

• 并发性,多线程

事件循环在线程中运行,通常是主线程,并在其线程执行所有回调和任务.
当一个任务在事件循环中运行时,没有其他任务可以在同一线程运行.当一个任务执行一个 await表达式时

正在运行的任务被挂起,事件循环执行下一个任务.

要调度来自另一个OS线程的callback,应该使用 loop.call_soon_threadsafe()方法. 例如

            loop.call_soon_threadsafe(callback, *args)

几乎所有异步对象 都不是线程安全的,这通常不是问题.

除法在任务或回调函数之外有代码 可以使用他们.如果需要这样的代码调用低级 异步API
应该使用loop.call_soon_threadsafe() 方法,如

            loop.call_soon_threadsafe(fut, cancel)

要从不同OS线程调度一个协程对象,应该使用run_coroutine_threadsafe() 函数.它返回一个 Future

            async def coro_func():
                    return await asyncio.sleep(1, 42)
            # Later in another OS thread
            future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
            # 等待结果
            result = future.result()

为了能够处理信号和执行子进程,事件循环必须运行于主线程中.

方法loop.run_in_executor() 可以和concurrent.future.ThreadPoolExecutor 一起使用.

用于在一个不同操作系统线程中执行阻塞代码,并避免阻塞运行事件循环的哪个操作系统线程.

目前没有其他办法能直接从另一个进程(例如通过 multiprocessing 启动的进程) 安排协程或回调.

事件循环方法 中 有一些 可以从管道读取并监视文件 描述符 而不会阻塞事件循环的API

此外,asyncio的子进程 API提供了一种 启动进程并从事件循环与其通信的办法.

最后,之前提到的 loop.run_inexecutor() 方法也可以配合 concurrent.futures.ProcessPoolExecutor

使用以在另一个进程 执行代码.

• 示例,不启用 debug = True时的运行时错误信息

  python asyncexample.py
     /asyncexample.py:16: RuntimeWarning: coroutine 'test' was never awaited
      test()
    RuntimeWarning: Enable tracemalloc to get the object allocation traceback
  

• 示例,启动 debug=True时 的运行错误信息

    python asyncexample.py
     /asyncexample.py:16: RuntimeWarning: coroutine 'test' was never awaited
    Coroutine created at (most recent call last)
      File "/asyncexample.py", line 21, in <module>
        a = asyncio.run(main11(),debug=True)
      File "/asyncio/runners.py", line 44, in run
        return loop.run_until_complete(main)
      File "/asyncio/base_events.py", line 629, in run_until_complete
        self.run_forever()
      File "/asyncio/base_events.py", line 596, in run_forever
        self._run_once()
      File "/asyncio/base_events.py", line 1882, in _run_once
        handle._run()
      File "/asyncio/events.py", line 80, in _run
        self._context.run(self._callback, *self._args)
      File "/asyncexample.py", line 16, in main11
        test()
      test()
    RuntimeWarning: Enable tracemalloc to get the object allocation traceback
  

3.2 运行阻塞的代码

不应该直接调用阻塞(CPU绑定代码).例如,如果一个函数执行1秒的CPU 密集型计算,那么所有并发任务和IO操作都延迟1秒.

可以用执行器在不同线程 甚至不同进程运行任务,以避免使用事件循环阻塞 OS 线程.
loop.run_in_executor() 了解详情.

3.3 日志记录

asyncio使用 logging模块,所有日志记录都是通过asyncio logger执行的
默认日志记录是 logging.INFO,可以很容易调整
logging.getLogger(“asyncio”).setLevel(logging.WARNING)

3.4 检测 never-awaited 协同程序

当协程函数被调用,而不是被等待时, 即执行 coro() 而不是 await coro() 或协程没有通过 asyncio.create_task() 被排入计划日程,asyncio将发出一条 RuntimWarning

import asyncio
async def test():
	print("never scheduled")
async def main():
	test()
asyncio.run(main())	
##
	test.py:7: RuntimeWarning:coroutine 'test' was never awaited
		test()

3.5 抛出异常 用户处理错误,而不是检测到错误退出

如果调用 Future.set_exception() 但不等的Future对象,将异常传播到用户代码.
这种情况看下,当Future对象被垃圾收集时,asyncio将发出一条日志消息.

async def bug():
	raise Exception("not consumed")

async def main():
    asyncio.create_task(bug())

未启用调试模式

python asyncexample.py
Task exception was never retrieved
future: <Task finished name='Task-2' coro=<bug() done, defined at /asyncexample.py:20> exception=Exception('not consumed')>
Traceback (most recent call last):
  File "/asyncexample.py", line 21, in bug
    raise Exception("not consumed")
Exception: not consumed

使用调试模式debug=True,以便跟踪信息

	python asyncexample.py
	Task exception was never retrieved
	future: <Task finished name='Task-2' coro=<bug() done, defined at /asyncexample.py:20> exception=Exception('not consumed') created at /asyncio/tasks.py:361>
	source_traceback: Object created at (most recent call last):
	  File "/asyncexample.py", line 30, in <module>
	    a = asyncio.run(main(),debug=True)
	  File "/asyncio/runners.py", line 44, in run
	    return loop.run_until_complete(main)
	  File "/asyncio/base_events.py", line 629, in run_until_complete
	    self.run_forever()
	  File "/asyncio/base_events.py", line 596, in run_forever
	    self._run_once()
	  File "/asyncio/base_events.py", line 1882, in _run_once
	    handle._run()
	  File "/asyncio/events.py", line 80, in _run
	    self._context.run(self._callback, *self._args)
	  File "/asyncexample.py", line 24, in main
	    asyncio.create_task(bug())
	  File "/asyncio/tasks.py", line 361, in create_task
	    task = loop.create_task(coro)
	Traceback (most recent call last):
	  File "/asyncexample.py", line 21, in bug
	    raise Exception("not consumed")
	Exception: not consumed

小结

我们这里介绍了 经典异步编程中一些需要理解的同步原语，并且举例实际使用时的步骤，和需要注意的问题。

异步编程可以大大提供程序的响应能力，特别是在实时系统和多个操作的业务中。

本节代码：

 github.com/hahamx/examples/tree/main/alg_practice/1_pys_async