自学 python 中的异步编程 asyncio (五)：asyncio 与线程thread网站首页 其他

自学 python 中的异步编程 asyncio (五)：asyncio 与线程thread

Eaton5959 2023-06-26 12:00:02

简介python 异步执行函数

1 asyncio 与线程thread 相结合

第一种得到异步运行

线程可以简单地理解为程序中的一个分支，它可以独立地执行一些操作
在这段代码中，我们创建了一个新的线程，并将一个新的事件循环（Event Loop）传递给它。
使用 asyncio.run_coroutine_threadsafe() 方法将三个异步协程注册到新的事件循环中，并在主线程中立即返回一个协程对象。

import asyncio
import time
from threading import Thread


def loop_in_thread(_new_loop):  # 一个将被丢进线程的函数
    asyncio.set_event_loop(_new_loop)  # 调用loop需要使用set_event_loop方法指定loop
    _new_loop.run_forever()  # run_forever() 会永远阻塞当前线程，直到有人停止了该loop为止。


async def func01(num):  # 一个协程
    await asyncio.sleep(2)
    print('异步任务{}花费时间：{}秒'.format(num, time.time() - now_time))
    return '异步任务{}完成时间：{}秒'.format(num, time.time() - now_time)


if __name__ == '__main__':
    now_time = time.time()  # 程序运行时的时间戳

    # 创建一个新的loop
    # get_event_loop()只会在主线程创建新的event loop，其他线程中调用 get_event_loop() 则会报错
    new_loop = asyncio.new_event_loop()

    t = Thread(target=loop_in_thread, args=(new_loop,))  # 创建线程
    t.start()  # 启动线程

    # 调用asyncio.run_coroutine_threadsafe实现回调
    even = asyncio.run_coroutine_threadsafe(func01(1), new_loop)
    even.cancel()

    # 注册到新的事件循环中
    asyncio.run_coroutine_threadsafe(func01(2), new_loop)
    asyncio.run_coroutine_threadsafe(func01(3), new_loop)

    print('主进程运行花费时长：{}秒'.format(time.time() - now_time))

第二种得到同步运行

import asyncio
import time
from threading import Thread


def loop_in_thread(_loop):  # 创建线程版洗衣机
    asyncio.set_event_loop(_loop)  # 在线程中调用loop需要使用set_event_loop方法指定loop
    _loop.run_forever()  # run_forever() 会永远阻塞当前线程，直到有人停止了该loop为止。


def func01(num):  # 定义同步函数，而非协程
    time.sleep(1)
    print('任务{}花费时间：{}秒'.format(num, time.time() - now_time))
    return '任务{}完成时间：{}秒'.format(num, time.time() - now_time)


if __name__ == '__main__':
    now_time = time.time()

    # 创建一个新的loop，get_event_loop()只会在主线程创建新的event loop，其他线程中调用 get_event_loop() 则会报错
    new_loop = asyncio.new_event_loop()

    t = Thread(target=loop_in_thread, args=(new_loop,))  # 创建线程
    t.start()  # 启动线程

    # 调用call_soon_threadsafe实现回调（详细描述往下找）
    even = new_loop.call_soon_threadsafe(func01, 1)

    new_loop.call_soon_threadsafe(func01, 2)
    new_loop.call_soon_threadsafe(func01, 3)

分析上面两段代码

它们的相同点是都创建了一个新的 asyncio 事件循环对象 new_loop
并使用 asyncio.set_event_loop() 方法将该事件循环设置为新线程的默认事件循环。
不同：
- ”同步“ 使用 call_soon_threadsafe() 方法将同步任务作为回调函数加入到事件循环中，使得任务可以被异步执行。这种方式需要使用 new_loop.call_soon_threadsafe() 方法在新线程中调度回调函数的执行，而不是通过 new_loop.run_forever() 方法让事件循环一直运行。
- “异步” 使用了 asyncio.run_coroutine_threadsafe() 方法将异步任务作为回调函数加入到事件循环中，使得任务可以被异步执行。该方法会返回一个 concurrent.futures.Future 对象，通过该对象可以随时取消协程的执行。
总之，这两种方法都可以在多线程场景下使用 asyncio，实现异步任务的执行。
1. 而第一种方式使用了协程，更符合异步编程的理念，代码更加简洁易懂。但是由于协程是依赖事件循环的，如果在主线程中不小心使用了阻塞操作，会导致整个事件循环阻塞，从而影响所有异步任务的执行。
2. 而第二种方式则没有这个问题，可以在主线程中继续执行其他任务。

2 事件循环的易混淆概念

现在说明一下 set_event_loop()、new_event_loop()、get_event_loop() 这三个方法的区别和使用场景

首先，这三个方法都是 asyncio 中用于管理事件循环的方法。
事件循环是异步编程中的核心概念，它负责在异步任务之间进行切换，从而实现高效的非阻塞式IO操作。

1 get_event_loop()

get_event_loop() 是用于获取当前事件循环的方法。

如果 当前线程中 已经有事件循环在运行，那么这个方法将返回该事件循环。
否则，它将创建一个新的事件循环并返回。
使用 get_event_loop() 方法可以方便地在程序中获取事件循环，而不需要手动创建。

下面是一个示例：

import asyncio

async def coro():
    await asyncio.sleep(1)
    print('coro')

    
async def main():
    loop = asyncio.get_event_loop()  # 直接获取已有的事件循环
    await loop.create_task(coro())  # 并在这个旧的事件循环中，创建任务并运行
    print('main')

    
if __name__ == '__main__':
    # 创建一个事件循环，并直接运行协程
    asyncio.run(main())

在这个示例中，get_event_loop() 方法用于获取当前事件循环。由于我们使用了 asyncio.run() 来运行程序，因此它会自动创建事件循环并运行我们的 main() 协程。在 main() 协程中，我们又使用了 get_event_loop() 方法来获取当前事件循环。在这个示例中，这两个调用都返回的是同一个事件循环对象。

2 new_event_loop() 与 set_event_loop()

new_event_loop() 方法是用于创建一个新的事件循环的方法。

new_event_loop()不会自动设置为当前事件循环，需要使用 set_event_loop() 方法来将它设置为当前事件循环。

注意

在使用 new_event_loop() 方法创建的新事件循环中，没有任何任务在运行
- 因此，在创建完事件循环后，需要手动将任务添加到事件循环中
一个线程只能有一个默认事件循环
- 如果在一个线程中多次调用 set_event_loop() 方法，后面的调用会覆盖前面的调用，即只有最后一次调用才会生效

下面是一个示例：

import asyncio

async def coro():
    await asyncio.sleep(1)
    print('coro')

async def main():
    new_loop = asyncio.new_event_loop()  # 创建一个新的事件循环
    asyncio.set_event_loop(new_loop)  # 设置new_loop为事件循环
    await loop.create_task(coro())  # 在new_loop里面创建任务，而不是外面的主事件循环
    print('main')

if __name__ == '__main__':
    # 创建一个事件循环，并直接运行协程
    asyncio.run(main())

在这个示例中，我们使用了 new_event_loop() 方法来创建一个新的事件循环
并使用 set_event_loop() 方法将它设置为当前事件循环。
然后，我们在这个事件循环中运行了一个协程。
需要注意的是，在使用 new_event_loop() 方法创建的新事件循环中，没有任何任务在运行。
- 因此，在创建完事件循环后，需要手动将任务添加到事件循环中。
需要注意的是，一个线程只能有一个默认事件循环。
- 如果在一个线程中多次调用 set_event_loop() 方法，后面的调用会覆盖前面的调用，即只有最后一次调用才会生效。