异步编程中你需要知道的 指令、响应,同步原语 和 使用时注意事项

简介: 【5月更文挑战第1天】`asyncio`提供了多种同步原语,如Lock、Event、Condition、Semaphore和BoundedSemaphore。Lock用于基本同步,Event用于标志状态切换,Condition允许在满足特定条件时唤醒协程,Semaphore控制资源访问,BoundedSemaphore防止计数超过预设值。

1 同步原语 Synchronization primitives

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

锁:

    Lock
    Event
    Condition

信号量:

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

1.1 锁具

1.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()

1.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 信号量 Semaphores

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 有界信号量

有界信号量实现,继承自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

4 小结

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

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

本节代码:

 github.com/hahamx/examples/tree/main/alg_practice/1_pys_async
相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
目录
相关文章
|
1月前
|
JavaScript 前端开发 UED
在 JavaScript 中,异步编程和回调函数是处理非阻塞操作(如网络请求、文件读写等)的重要工具
【5月更文挑战第10天】JavaScript中的异步编程和回调函数用于处理非阻塞操作,提高应用响应性和吞吐量。异步编程通过回调函数、Promises和async/await等方式实现,避免程序因等待操作完成而阻塞。回调函数是异步操作完成后调用的函数,常用于处理网络请求等。然而,回调函数存在嵌套问题和错误处理困难,因此出现了Promises和async/await等更优解决方案。
18 3
|
2天前
|
前端开发 JavaScript 数据处理
在JavaScript中,异步函数是指那些不会立即执行完毕,而是会在未来的某个时间点(比如某个操作完成后,或者某个事件触发后)才完成其执行的函数
【6月更文挑战第15天】JavaScript中的异步函数用于处理非同步任务,如网络请求或定时操作。它们使用回调、Promise或async/await。
14 7
|
1月前
|
存储 前端开发 JavaScript
面试官问:如果有100个请求,你如何使用Promise控制并发?
面试官问:如果有100个请求,你如何使用Promise控制并发?
|
1月前
|
前端开发
如何处理前端应用程序中的异步操作
前端异步操作常见方法包括回调函数、Promise 和 async/await。回调函数可能导致回调地狱,Promise 提供了更好的错误处理和链式调用,而 async/await 则基于 Promise,以同步风格处理异步任务,提高代码可读性。根据需求和喜好选择相应方法,以实现更可靠、易维护的代码。
|
1月前
|
开发框架 JavaScript 前端开发
描述JavaScript事件循环机制,并举例说明在游戏循环更新中的应用。
JavaScript的事件循环机制是单线程处理异步操作的关键,由调用栈、事件队列和Web APIs构成。调用栈执行函数,遇到异步操作时交给Web APIs,完成后回调函数进入事件队列。当调用栈空时,事件循环取队列中的任务执行。在游戏开发中,事件循环驱动游戏循环更新,包括输入处理、逻辑更新和渲染。示例代码展示了如何模拟游戏循环,实际开发中常用框架提供更高级别的抽象。
18 1
|
1月前
|
缓存 小程序 前端开发
如何解决小程序异步请求问题
如何解决小程序异步请求问题
126 0
|
8月前
|
监控 Go
Go 语言一次性定时器使用方式和实现原理
Go 语言一次性定时器使用方式和实现原理
45 0
|
前端开发 JavaScript API
浏览器原理 19 # JavaScript 引擎是如何实现 async / await 以同步的方式来编写异步代码的?
浏览器原理 19 # JavaScript 引擎是如何实现 async / await 以同步的方式来编写异步代码的?
139 0
浏览器原理 19 # JavaScript 引擎是如何实现 async / await 以同步的方式来编写异步代码的?
多种网络请求方式 ,这么骚气的操作确定不来看看嘛?
兄弟们,网络真是个好东西,里面去个个都是人才,说话又好听,超喜欢在里面。 确实,网络已经是我们日常生活中密不可分的一部分了 那咱们今天就来讲讲Flutter里的网络。
多种网络请求方式 ,这么骚气的操作确定不来看看嘛?
|
移动开发 JavaScript 前端开发
前端开发面试题—JavaScript执行机制(同步与异步,补充:线程与进程)
今天分享一下我遇到的一个面试题,是关于JavaScript执行机制——同步与异步的问题,解释一下什么是同步和异步呢?
197 0
前端开发面试题—JavaScript执行机制(同步与异步,补充:线程与进程)