Swagger生成的Connexion-Tornado服务如何实现部分接口异步处理

原有架构基于Connexion+Tornado的组合，通过WSGIContainer桥接Swagger生成的WSGI服务，因为WSGI是同步接口，长耗时接口（约30秒）会阻塞主事件循环，导致后续请求排队。官方文档的说明如下：

WSGI是同步接口，而Tornado的并发模型基于单线程异步执行。这意味着使用Tornado的WSGIContainer运行WSGI应用的可扩展性，低于在gunicorn、uWSGI这类多线程WSGI服务器中运行同一应用的效果。

之前尝试的两套改造方案均存在实现层面的缺陷，具体问题和修正方案如下：

之前两次改造失败的核心原因 #

• 第一套AsyncFallbackHandler实现的问题
  • 事件循环报错根源：原代码在工作线程内尝试获取当前线程绑定的事件循环，而ThreadPoolExecutor的工作线程默认无关联事件循环，直接抛出RuntimeError: There is no current event loop in thread 'ThreadPoolExecutor-0_0'
  • 未正确实现WSGI规范的桥接逻辑：WSGI应用需要通过start_response回调返回状态码、响应头，原代码直接调用WSGI入口但未收集返回的响应数据，相当于只执行了业务逻辑但没把结果回传给客户端
  • 破坏Tornado内部状态：手动设置self._finished = True、直接调用on_finish()属于操作类内部私有属性，会导致响应流程紊乱
• 第二套自定义WifiRequestHandler实现的问题
  • 参数传递错误：提交到线程池时把完整的args元组作为单个参数传给业务方法，而非解包传参
  • 响应对象处理不当：ConnexionResponse是继承自Werkzeug的Response类，并非可直接序列化的字典/字符串，原代码直接尝试用JSON编码器序列化该对象，自然会报错
  • 未正确桥接响应管道：Tornado的write方法只接受字节、字符串、字典类型的输入，不能直接传入Response对象，需要拆解其内部的状态码、响应头、响应体后再回写

可落地的改造实现 #

核心思路：不绕开Connexion的原生路由、参数校验、响应序列化能力，仅将同步的WSGI调用提交到独立线程池执行，全程不阻塞Tornado主事件循环，同时正确桥接WSGI响应到Tornado的响应管道。

第一步：改造异步WSGI回退处理器 #

替换原有错误的AsyncFallbackHandler实现，代码如下：

import concurrent.futures
import tornado.web
import tornado.wsgi

class AsyncFallbackHandler(tornado.web.RequestHandler):
    # 线程池大小根据业务并发量、接口耗时调整，30秒长耗时接口建议配置15~20个线程
    _executor = concurrent.futures.ThreadPoolExecutor(max_workers=20)

    def initialize(self, fallback: tornado.wsgi.WSGIContainer):
        self.wsgi_container = fallback

    def _execute_sync_wsgi(self):
        # 同步执行逻辑，会被提交到线程池运行，不阻塞主IOLoop
        # 构造符合WSGI规范的请求环境
        wsgi_env = tornado.wsgi.WSGIContainer.environ(self.request)
        response_meta = {}

        def start_response(status, headers, exc_info=None):
            # 收集WSGI返回的状态码、响应头
            response_meta['status_code'] = int(status.split(' ')[0])
            response_meta['headers'] = headers

        # 执行WSGI应用逻辑，收集响应体
        response_body = b''.join(self.wsgi_container(wsgi_env, start_response))
        return response_meta, response_body

    async def prepare(self):
        # 将同步WSGI执行逻辑提交到线程池，等待执行完成
        response_meta, response_body = await self.run_in_executor(
            self._executor,
            self._execute_sync_wsgi
        )
        # 将收集到的响应数据正确写入Tornado响应
        self.set_status(response_meta['status_code'])
        for header_name, header_value in response_meta['headers']:
            self.set_header(header_name, header_value)
        self.write(response_body)
        self.finish()

第二步：修正自定义业务处理器的ConnexionResponse处理逻辑 #

如果保留单独的WifiRequestHandler处理wifi相关接口，只需修正ConnexionResponse的拆解逻辑，代码如下：

async def run_in_executor(self, method, *args):
    # 修正参数传递bug：解包args后再传给业务方法
    result = await self.run_in_executor(None, method, *args)
    # 单独处理ConnexionResponse类型的返回值
    if isinstance(result, ConnexionResponse):
        return {
            "status": result.status_code,
            "headers": dict(result.headers),
            "body": result.response[0] if result.response else b""
        }
    # 普通返回值走原有JSON编码逻辑
    return encoder.JSONEncoder().encode(result)

在回写响应时，对拆解后的ConnexionResponse数据单独处理状态码、响应头、响应体即可，不会再出现编码错误。

注意事项 #

• 线程池大小需要匹配实际业务并发能力：长耗时接口的执行全程在线程池内，若线程数不足，仍然会出现请求排队的问题
• 主事件循环不会被阻塞：所有同步业务逻辑都在线程池运行，Tornado主IOLoop可以正常处理WebSocket、接口调度等轻量逻辑
• 不需要完全弃用WSGIContainer：只要做好线程池隔离，既可以保留Connexion的Swagger生态能力，又能解决同步阻塞的问题

内容的提问来源于stack exchange，提问作者Straticiuc Vicu