Python非阻塞后台任务实现方法

golang学习网今天将给大家带来《在Python中利用asyncio和threading实现非阻塞后台任务 》，感兴趣的朋友请继续看下去吧！以下内容将会涉及到等等知识点，如果你是正在学习文章或者已经是大佬级别了，都非常欢迎也希望大家都能给我建议评论哈~希望能帮助到大家！

异步协程与线程的挑战

在构建基于asyncio的异步应用时，例如一个WebSocket服务器，我们经常会遇到需要执行一些持续性后台任务的需求，比如从消息队列（如SQS）接收数据并发送给客户端。一个常见的误区是尝试直接将一个异步函数（async def定义的函数）作为threading.Thread的目标函数来运行。

考虑以下示例代码片段：

import socketio
import threading
import json
import asyncio # 假设已导入

# ... (sio, app, connect, disconnect, item_removed等定义)

async def background_task():
    queue = SQSQueue() # 假设 SQSQueue 是一个同步或异步的SQS客户端
    while True:
        message = queue.get_next_message_from_sqs() # 假设这是一个阻塞或异步IO操作
        data = json.loads(message.body)
        await sio.emit('item_added', data) # 这是一个异步IO操作

background_thread = threading.Thread(target=background_task) # 错误的使用方式
background_thread.daemon = True
background_thread.start()

当以这种方式运行代码时，你可能会遇到RuntimeWarning: coroutine 'background_task' was never awaited.这样的警告。这是因为async def定义的函数在被调用时，并不会立即执行其内部逻辑，而是返回一个“协程对象”（coroutine object）。要真正执行协程内部的await操作，这个协程对象必须在一个asyncio事件循环中被“等待”（awaited）。threading.Thread的target参数期望的是一个普通的同步可调用对象，它并不会自动为异步函数创建并运行事件循环。因此，background_task协程对象被创建了，但从未进入事件循环执行，导致了警告。

解决方案：使用 asyncio.run 在独立线程中运行协程

解决上述问题的关键在于，确保在后台线程中为你的异步任务提供一个独立的asyncio事件循环。asyncio.run()函数正是为此设计的，它提供了一个运行协程的便捷方式：它会自动创建一个新的事件循环，运行传入的协程直到完成，然后关闭事件循环。

因此，正确的做法是将asyncio.run作为线程的目标函数，并将你的异步任务作为参数传递给它。

import socketio
import threading
import json
import asyncio # 确保导入 asyncio

# ... (sio, app, connect, disconnect, item_removed等定义)

async def background_task():
    # 假设 SQSQueue 是一个同步或异步的SQS客户端
    # 如果 get_next_message_from_sqs 是一个同步阻塞调用，它将阻塞此线程的事件循环
    # 如果是异步调用，则需要 await
    queue = SQSQueue() 
    while True:
        # 确保这里的IO操作是异步的，或者是非阻塞的
        # 如果是同步阻塞的IO，会阻塞整个线程
        message = queue.get_next_message_from_sqs() 
        data = json.loads(message.body)
        await sio.emit('item_added', data)
        # 可以在这里添加一个短暂的异步等待，避免CPU空转，例如：
        # await asyncio.sleep(0.1)

# 关键的修改在这里：
# 将 asyncio.run 作为线程的目标，并将 background_task 协程作为其参数
background_thread = threading.Thread(target=asyncio.run, args=(background_task(),)) # 注意双括号和尾部逗号
background_thread.daemon = True # 设置为守护线程，主程序退出时自动结束
background_thread.start()

请注意args=(background_task(),)中的双括号和尾部逗号。background_task()是调用background_task协程函数并获取其返回的协程对象，而(background_task(),)则是一个包含这一个协程对象的元组。threading.Thread的args参数期望一个可迭代对象（通常是元组），用于传递给target函数。

工作原理深入

当background_thread.start()被调用时：

1. 一个新的操作系统线程被创建。
2. 在这个新线程中，asyncio.run(background_task())被执行。
3. asyncio.run()在该线程内部：
   • 创建一个全新的asyncio事件循环。
   • 将background_task()返回的协程对象提交到这个新的事件循环中运行。
   • 开始运行这个事件循环，直到background_task协程完成（在我们的while True循环中，它永远不会“完成”，而是持续运行）。
   • background_task协程内部的所有await调用（如await sio.emit(...)）都会在这个新的、独立的事件循环中正确地调度和执行，而不会阻塞主线程中Uvicorn或Socket.IO服务器的事件循环。

这种方法允许你在不干扰主应用事件循环的情况下，在后台独立地运行asyncio协程。

注意事项与最佳实践

1. I/O密集型任务适用：此模式特别适合I/O密集型任务，因为asyncio的优势在于高效地处理并发I/O。如果你的后台任务是CPU密集型的，它仍然会阻塞整个线程，此时可能需要考虑使用multiprocessing模块来利用多核CPU。
2. 事件循环隔离：每个调用asyncio.run()的线程都会有其独立的事件循环。这意味着这些事件循环之间是隔离的，它们不会共享同一个事件循环上下文。
3. 资源共享与同步：如果后台线程的协程需要与主线程或其他线程共享数据或资源，必须小心处理并发访问，使用asyncio.Lock、threading.Lock或其他同步原语来避免竞态条件。
4. 优雅关闭：由于background_task是一个while True循环，它会无限运行。在生产环境中，你需要考虑如何优雅地停止这个后台线程，例如通过设置一个共享的事件标志（threading.Event或asyncio.Event），或者捕获信号来通知线程退出循环。
5. 异步SQS客户端：在background_task中，queue.get_next_message_from_sqs()的实现非常关键。如果它是一个同步阻塞调用，即使在异步函数中，它也会阻塞整个后台线程。理想情况下，你应该使用一个异步的SQS客户端（如aiobotocore）并await其方法，以充分发挥asyncio的非阻塞特性。

总结

通过将asyncio.run()作为threading.Thread的目标函数，并传入你的异步协程，你可以有效地在Python中创建非阻塞的后台任务。这种模式允许异步操作在独立的线程中拥有自己的事件循环，从而避免了主事件循环的阻塞，确保了应用程序的响应性和并发性，是处理WebSocket服务器等需要持续后台数据处理场景的强大工具。正确理解和应用asyncio与threading的结合，是构建高性能、可扩展Python异步应用的基石。

今天带大家了解了的相关知识，希望对你有所帮助；关于文章的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~