Skip to content

26.协程-事件循环EventLoop.md

Latest commit

 

History

History
373 lines (234 loc) · 12.7 KB

26.协程-事件循环EventLoop.md

File metadata and controls

373 lines (234 loc) · 12.7 KB
Raw

事件循环EventLoop

事件循环（Event Loop），通常简称为 event loop，是异步编程中的一个核心概念。它是一个持续运行的循环，用于处理和调度异步任务（例如协程）的执行，以确保它们按照预期的顺序和时间执行。事件循环允许多个任务并发运行，而无需阻塞主程序的执行。

一.事件循环的创建、获取和设置

asyncio.get_running_loop(): 获取当前正在运行的事件循环

asyncio.get_event_loop(): 获取或创建事件循环

asyncio.set_event_loop(loop): 设置当前线程的默认事件循环

asyncio.new_event_loop(): 创建一个新的事件循环

二.运行和停止事件循环

loop.run_until_complete(future): 运行事件循环，直到future运行结束

loop.stop(): 停止事件循环

loop.is_running(): 如果事件循环依然在运行，则返回True

loop.is_closed(): 如果事件循环已经close，则返回True

loop.close(): 关闭事件循环

三.创建Future和Task

loop.create_future(coroutine): 返回future对象

loop.create_task(corootine): 返回task对象

1.loop.set_task_factory

结构：

loop.set_task_factory(factory)

作用：

用于设置事件循环的任务工厂，任务工厂决定了在事件循环中创建任务时所使用的任务类，任务工厂是一个回调函数，它接受一个参数（通常是一个事件循环对象），并返回一个用于创建任务的函数或类。通过设置任务工厂，您可以自定义任务的创建过程。

例如，您可以使用自定义的任务类来扩展任务的功能。

import asyncio


def custom_task_factory(loop, coro):
    task = asyncio.Task(coro, loop=loop)
    task.my_custom_attribute = "This is a custom attribute"
    return task


async def my_coroutine():
    await asyncio.sleep(1)


async def main():
    loop = asyncio.get_event_loop()

    # 设置自定义任务工厂
    loop.set_task_factory(custom_task_factory)

    # 创建协程任务
    task = asyncio.create_task(my_coroutine())
    await task

    # 访问自定义属性
    print(task.my_custom_attribute)


asyncio.run(main())

在上述示例中，我们首先定义了一个自定义的任务工厂函数 custom_task_factory。该函数接受事件循环对象和协程对象作为参数，并使用 asyncio.Task 来创建任务。然后，我们在任务上添加了一个自定义属性 my_custom_attribute。

接着，我们在 main() 函数中获取事件循环，然后使用 loop.set_task_factory() 来设置自定义的任务工厂。这意味着以后通过 asyncio.create_task() 创建的任务将使用我们的自定义工厂函数。

最后，我们创建了一个协程任务 task，等待它完成，并访问了自定义属性 my_custom_attribute。这个自定义属性是通过任务工厂添加的。

2.loop.get_task_factory()

结构：

current_task_factory = asyncio.get_event_loop().get_task_factory()

作用：

获取当前事件循环的任务工厂。这是因为任务工厂通常是与事件循环关联的而不是与任务本身关联的。

import asyncio

def custom_task_factory(loop, coro):
    task = asyncio.Task(coro, loop=loop)
    task.my_custom_attribute = "This is a custom attribute"
    return task

async def my_coroutine():
    await asyncio.sleep(1)

async def main():
    loop = asyncio.get_event_loop()

    # 设置自定义任务工厂
    loop.set_task_factory(custom_task_factory)

    # 获取当前事件循环的任务工厂
    current_task_factory = loop.get_task_factory()
    print("Current task factory:", current_task_factory)

asyncio.run(main())

在上述示例中，我们首先设置了一个自定义的任务工厂 custom_task_factory，该工厂会在创建任务时添加一个自定义属性。然后，我们在 main() 函数中使用 loop.set_task_factory() 来设置自定义任务工厂。

接着，我们使用 loop.get_task_factory() 来获取当前事件循环的任务工厂，并将其存储在 current_task_factory 变量中。最后，我们在输出中打印了当前任务工厂。

注意：

如果未设置自定义任务工厂，则 loop.get_task_factory() 将返回默认的任务工厂。

设置自定义任务工厂可以让您在创建任务时自定义任务对象的行为，例如添加自定义属性、记录日志等。

四.事件时钟-time

结构：

loop = asyncio.get_event_loop()
current_time = loop.time()

返回值：

返回一个浮点数，表示事件循环内部的时间。这个时间戳通常用于测量异步任务的执行时间或在协程中实现超时操作。您可以在协程内部使用这个时间戳来计算时间间隔、执行时间等

作用：

用于获取事件循环的时间戳。时间戳表示了事件循环内部的时钟时间，通常以秒为单位。

import asyncio

async def my_coroutine():
    start_time = asyncio.get_event_loop().time()
    await asyncio.sleep(1)
    end_time = asyncio.get_event_loop().time()
    elapsed_time = end_time - start_time
    print(f"Coroutine took {elapsed_time:.2f} seconds to complete")

async def main():
    await my_coroutine()

if __name__ == "__main__":
    asyncio.run(main())

五.计划执行回调函数(CallBacks)

1.loop.call_later

结构：

loop = asyncio.get_event_loop()
loop.call_later(delay, callback, *args, context=None)

参数：

delay: 延迟的时间，以秒为单位，指定了回调函数将在多长时间之后执行。

callback: 要调度执行的回调函数，可以是一个普通函数、异步函数、协程函数或任何可调用对象。

*args: 可选的位置参数，这些参数将传递给回调函数。

context: 可选的上下文对象，可以用于传递额外的信息给回调函数。

作用：

在指定的一段时间之后调度执行一个函数（回调函数）。这个方法允许您安排在未来的某个时间点执行某个操作，通常用于实现延迟执行的功能。

import asyncio

def my_callback(arg1, arg2):
    print(f"Callback called with arguments: {arg1}, {arg2}")

async def main():
    loop = asyncio.get_event_loop()
    loop.call_later(2, my_callback, "Hello", "World")
    await asyncio.sleep(3)  # 等待足够的时间以确保回调函数执行

if __name__ == "__main__":
    asyncio.run(main())

注意

loop.call_later() 返回一个定时器对象，可以用于取消计划的回调函数执行。

如果需要取消计划的回调，可以使用 loop.call_later() 返回的定时器对象的 cancel() 方法。

2.loop.call_at

结构:

loop = asyncio.get_event_loop()
loop.call_at(when, callback, *args, context=None)

参数:

when: 表示绝对时间的浮点数，以秒为单位，指定了回调函数将在何时执行。

callback: 要调度执行的回调函数，可以是一个普通函数、异步函数、协程函数或任何可调用对象。

*args: 可选的位置参数，这些参数将传递给回调函数。

context: 可选的上下文对象，可以用于传递额外的信息给回调函数。

作用：

在指定的绝对时间点调度执行一个函数（回调函数）。与 loop.call_later() 不同，loop.call_at() 使用绝对时间而不是相对时间来安排回调函数的执行。这允许您精确地安排回调函数在未来的某个特定时间点执行。

import asyncio

def my_callback(arg1, arg2):
    print(f"Callback called with arguments: {arg1}, {arg2}")

async def main():
    loop = asyncio.get_event_loop()
    current_time = loop.time()  # 获取当前时间戳

    # 计算未来的绝对时间点（例如，2 秒后）
    target_time = current_time + 2

    # 使用 loop.call_at() 安排回调函数在未来的特定时间点执行
    loop.call_at(target_time, my_callback, "Hello", "World")

    await asyncio.sleep(3)  # 等待足够的时间以确保回调函数执行

if __name__ == "__main__":
    asyncio.run(main())

注意:

loop.call_at() 返回一个定时器对象，可以用于取消计划的回调函数执行。如果需要取消计划的回调，可以使用 loop.call_at() 返回的定时器对象的 cancel() 方法。

3.loop.call_soon

结构：

loop = asyncio.get_event_loop()
loop.call_soon(callback, *args, context=None)

参数：

callback: 要调度执行的回调函数，可以是一个普通函数、异步函数、协程函数或任何可调用对象。

*args: 可选的位置参数，这些参数将传递给回调函数。

context: 可选的上下文对象，可以用于传递额外的信息给回调函数。

作用：

用于调度尽快执行一个函数（回调函数）。这个方法将回调函数放入事件循环队列中，以便在事件循环空闲时立即执行。通常情况下，它用于确保某个函数能够尽快被调用，而不需要等待事件循环的下一个迭代。

import asyncio

def my_callback(arg1, arg2):
    print(f"Callback called with arguments: {arg1}, {arg2}")

async def main():
    loop = asyncio.get_event_loop()
    
    # 使用 loop.call_soon() 调度回调函数立即执行
    loop.call_soon(my_callback, "Hello", "World")

    await asyncio.sleep(0.1)  # 等待一小段时间以确保回调函数执行

if __name__ == "__main__":
    asyncio.run(main())

注意:

回调函数的执行顺序可能受到事件循环中其他任务的影响，因此不能保证立即执行。如果需要更精确的调度，可以使用 loop.call_later() 或 loop.call_at() 来安排延迟或绝对时间点的执行。

4.loop.call_soon_threadsafe

结构：

loop = asyncio.get_event_loop()
loop.call_soon_threadsafe(callback, *args, context=None)

参数：

callback: 要调度执行的回调函数，可以是一个普通函数、异步函数、协程函数或任何可调用对象。

*args: 可选的位置参数，这些参数将传递给回调函数。

context: 可选的上下文对象，可以用于传递额外的信息给回调函数。

作用：

它类似于 loop.call_soon()，用于在事件循环中调度执行一个函数（回调函数）。但与 loop.call_soon() 不同的是，loop.call_soon_threadsafe() 可以在多线程环境中安全地调用，以确保异步操作不会引发线程安全问题。这对于在多线程应用程序中与事件循环进行交互非常有用，例如，从不同的线程调度异步任务。

import asyncio
import threading

def my_callback(arg1, arg2):
    print(f"Callback called with arguments: {arg1}, {arg2}")

async def main():
    loop = asyncio.get_event_loop()

    # 在不同的线程中调用 loop.call_soon_threadsafe()
    thread = threading.Thread(target=lambda: loop.call_soon_threadsafe(my_callback, "Hello", "World"))
    thread.start()
    thread.join()

if __name__ == "__main__":
    asyncio.run(main())

在这个示例中，我们首先定义了一个回调函数 my_callback，然后在 main() 协程内部使用 loop.call_soon_threadsafe() 安全地调度回调函数执行。在不同的线程中，我们创建了一个新线程，并在该线程内部使用 loop.call_soon_threadsafe() 调度回调函数执行。这样可以确保在多线程环境中与事件循环进行安全的交互。

5.注意

上面的几个回调函数都只使用了“位置参数”哦，asyncio中，大部分的计划回调函数都不支持“关键字参数”，如果是想要使用关键字参数，则推荐使用functools.aprtial()对方法进一步包装。

import functools

def power(base, exponent):
    return base ** exponent

# 使用 functools.partial 包装 power 函数，固定 base 参数为 2
square = functools.partial(power, base=2)

# 调用 square 函数，只需传递 exponent 参数
result = square(exponent=3)
print(result)  # 输出 8

6.总结

CallBack函数只能够定义为同步方法，不能够定义为异步方法。

每一个CallBack方法只会调用一次，如果在同一个时刻有另个CallBack方法需要调用，则他们的执行顺序是不确定的。

注意使用functools.partial（）去修饰带有关键字参数的CallBack方法；

如何理解？

对于一般的异步函数，我们需要将它放在事件循环里面，然后通过事件循环去循环调用它，而因为CallBack并不是异步函数，它是定义为普通的同步方法，所以不能够放在事件循环里面，但是如果我依然想要让事件循环去执行它怎么办呢？那就不放进事件循环，直接让事件循环“立即、稍后、在什么时候”去执行它不就行了嘛，call的含义就是“执行”。