Python asyncio应用中后台协程任务的正确运行姿势

import socketio
import threading
import json
# from sqs_handler import SQSQueue # 假设这是一个处理SQS的类

sio = socketio.AsyncServer(async_mode='asgi')
app = socketio.ASGIApp(sio, static_files={"/": "./"})

@sio.event
async def connect(sid, environ):
    print(sid, "connected")

@sio.event
async def disconnect(sid):
    print(sid, "disconnected")

async def background_task():
    # queue = SQSQueue()
    print("后台任务启动，等待SQS消息...")
    while True:
        # message = queue.get_next_message_from_sqs() # 假设这里会阻塞或需要await
        # data = json.loads(message.body)
        # await sio.emit('item_added', data)
        # 模拟异步操作
        await asyncio.sleep(1) # 模拟从SQS获取消息的延迟
        print("后台任务正在运行...")
        await sio.emit('item_added', {'message': 'New item added from background!'})

# 错误的方式：直接将协程对象作为线程目标
# background_thread = threading.Thread(target=background_task)
# background_thread.daemon = True
# background_thread.start()

import socketio
import threading
import json
import asyncio # 1. 导入 asyncio

# 假设 SQSQueue 能够以非阻塞方式获取消息，或者其内部处理了阻塞
# 为了演示，我们创建一个模拟的 SQSQueue
class SQSQueue:
    def get_next_message_from_sqs(self):
        # 模拟阻塞操作，实际应用中应使用异步 SQS 客户端
        import time
        time.sleep(0.5)
        return type('obj', (object,), {'body': json.dumps({'id': 123, 'status': 'added', 'data': 'example'})})()

sio = socketio.AsyncServer(async_mode='asgi')
app = socketio.ASGIApp(sio, static_files={"/": "./"})

@sio.event
async def connect(sid, environ):
    print(f"客户端 {sid} 已连接")

@sio.event
async def disconnect(sid):
    print(f"客户端 {sid} 已断开")

@sio.event
async def item_removed(sid, data):
    print(f"收到 item_removed 事件，数据: {data}")
    await sio.emit("item_removed", data)

async def background_task():
    """
    后台任务：从SQS接收消息并发送给所有连接的客户端。
    此协程将在独立的事件循环中运行。
    """
    queue = SQSQueue()
    print("后台任务启动，等待SQS消息...")
    while True:
        try:
            # 注意：如果 get_next_message_from_sqs 是阻塞的，它会阻塞此线程的事件循环。
            # 理想情况下，SQSQueue 应该使用异步客户端（如 aiobotocore）并提供 awaitable 方法。
            message = queue.get_next_message_from_sqs()
            data = json.loads(message.body)
            print(f"从SQS接收到消息: {data}，准备发送给客户端")
            await sio.emit('item_added', data)
            await asyncio.sleep(0.1) # 避免忙等待，给事件循环调度其他任务的机会
        except Exception as e:
            print(f"后台任务出错: {e}")
            await asyncio.sleep(1) # 错误后等待一段时间重试

# 2. 关键修改：使用 asyncio.run 作为线程目标，并将协程作为 args 传递
background_thread = threading.Thread(target=asyncio.run, args=(background_task,))
background_thread.daemon = True # 设置为守护线程，主程序退出时自动终止
background_thread.start()
print("后台线程已启动。")

# 运行 ASGI 应用 (例如使用 uvicorn)
# if __name__ == "__main__":
#     import uvicorn
#     print("主应用启动...")
#     uvicorn.run(app, host="0.0.0.0", port=8000)