Python同时运行web.run_app()与其他异步程序的报错问题咨询

两个报错的根本原因如下：

• Type1报错：web.run_app()是aiohttp提供的快捷启动方法，内部会自动创建并运行一个新的事件循环，而你主线程已经通过asyncio.run(main())启动了一个事件循环，同一个线程内不能同时运行两个事件循环，因此抛出循环已运行的异常。
• Type2报错：web.run_app()默认会注册系统信号处理器用于优雅停机，而系统信号的处理逻辑只能在主解释器的主线程中运行，放到子线程启动就会触发set_wakeup_fd相关错误。

正确的解决思路是放弃使用web.run_app()，改用aiohttp更底层的服务启动接口，把socketio服务挂载到当前已经运行的主线程事件循环中，不需要新开线程也不需要新开事件循环，同时修复原有代码中的其他异步阻塞问题。

import os, sys
import asyncio
import platform
import socketio
import json

from aioconsole import ainput
from aiohttp import web
from array import *

Result = -1 
Inputs_Required = True 
Input_arr = array('i')

sio = socketio.AsyncServer() 
app = web.Application() 
sio.attach(app)
Host = "192.168.0.7" 
Port = 8050        

async def IOBlock():
    global Input_arr
    global Inputs_Required
    
    while(True):
        response = await ainput("Enter new input? (y/n): ")
        if('y' == response or 'Y' == response):
            Input = await ainput("Enter number to be computed: ")
            Input_arr.append(int(Input))
            break
        elif('n' == response or 'N' == response):
            Inputs_Required = False
            break
        else:
            print("Invalid response.")

async def main():
    global Result
    global Inputs_Required
    global Input_arr
    
    WebSocketStarted = False
            
    try:
        while True: 
            if(Inputs_Required == True):
                Task_AddInput = asyncio.create_task(IOBlock())
                await Task_AddInput
            elif (WebSocketStarted == False):
                WebSocketStarted = True
                # 直接创建web服务后台任务，不需要await避免阻塞后续逻辑
                asyncio.create_task(WebStreaming())
            
            if(len(Input_arr) > 0):
                Task_PrintResult = asyncio.create_task(EvaluateResult()) 
                await Task_PrintResult
                
    except Exception as x:
        print(x)
    finally: 
        await Cleanup()

async def WebStreaming():
    print("Starting web-socket streaming of sensor data..")
    # 用底层接口启动web服务，复用当前事件循环
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, host=Host, port=Port)
    await site.start()
    # 保持服务持续运行
    while True:
        await asyncio.sleep(3600)
    
async def EvaluateResult():
    global Input_arr
    global Result
    
    Result = 0
    for i in range (0, len(Input_arr)):
        Result += Input_arr[i]
    print(f"The sum of inputs fed so far = {Result}.")
    await asyncio.sleep(5)

async def Cleanup():
    global Input_arr
    global Inputs_Required
    global Result
    
    print("Terminating program....")
    Result = -1
    Inputs_Required = True
    for i in reversed(range(len(Input_arr))):
        del Input_arr[i]

@sio.event
async def connect(sid, environ):
    print("connect ", sid) 
    
@sio.event
async def OnClientMessageReceive(sid, data):
    global Result
    
    print("Client_message : ", data)
    while True: 
        msg = json.dumps(Result)
        print(msg)
        # 用emit发送自定义事件，指定接收客户端sid
        await sio.emit('OnServerMessageReceive', msg, room=sid)
        # 避免推送频率过高占用过多资源
        await asyncio.sleep(1)

@sio.event
def disconnect(sid):
    print('disconnect ', sid)

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

• 替换IOBlock中的同步input()为异步ainput()，避免同步输入阻塞整个事件循环，导致websocket服务无法响应
• 移除所有子线程、新开事件循环的相关逻辑，改用AppRunner + TCPSite启动web服务，直接复用当前主线程的事件循环，从根源避免两个报错
• 修复socketio自定义事件发送逻辑：将sio.send()改为sio.emit()，send()仅用于发送默认的message事件，自定义事件需要用emit，同时添加sid参数指定接收客户端，添加推送间隔避免资源占用过高
• 调整主任务逻辑，将web服务作为后台异步任务运行，不会阻塞后续的输入采集、计算逻辑，两个异步流程并行执行
• 修复原有代码的语法错误：asyncio.new_event_loop缺少括号、main()中未定义的Results全局变量拼写错误

内容的提问来源于stack exchange，提问作者Belmannu Raviraj Acharya