使用龙卷风大摇大摆生成服务器的异步端点

Question

我已经使用 swagger 编辑器生成了一个服务器。然后我开始使用 Tornado 作为 http-server，例如：

def main():
    app = App(system_manager=system_manager, import_name=__name__,
              specification_dir='./swagger/', server='tornado')
    app.app.json_encoder = encoder.JSONEncoder
    app.add_api('swagger.yaml', arguments={
                'title': 'API'}, pythonic_params=True)
    app.run(port=8085)

应用在哪里：

class App(connexion.App):
def __init__(self, system_manager, import_name, server='tornado', **kwargs):
    super(App, self).__init__(import_name, server=server, **kwargs)
    if not issubclass(type(system_manager), SystemManager):
        raise ValueError(
            "App.init: 'system_manager' is not a subclass of 'SystemManager'")
    self.__system_manager = system_manager

def run(self, port=None, server=None, debug=None, host=None, **options):
    server_type = server or self.server
    if server_type != 'tornado':
        super(App, self).run(port=port, server=server,
                             debug=debug, host=host, **options)
        return None
    if port is not None:
        self.port = port
    elif self.port is None:
        self.port = 5000

    self.host = host or self.host or '0.0.0.0'

    if server is not None:
        self.server = server

    if debug is not None:
        self.debug = debug

    wsgi_container = tornado.wsgi.WSGIContainer(self.app)
    http_server = tornado.web.Application([
        (r'/websocket/.*', WebSocket, dict(system_manager=self.__system_manager)),
        (r'^/v1/wifi(/all)*$', AsyncFallbackHandler,
         dict(fallback=wsgi_container)),
        (r'.*', tornado.web.FallbackHandler, dict(fallback=wsgi_container))
    ], websocket_ping_interval=5)
    http_server.listen(self.port, address=self.host)
    tornado.ioloop.IOLoop.instance().start()

由于某些原因，我有一些端点需要大约 30 秒才能响应，并且因为我使用的是 WSGIContainer，所有请求都是同步的。这意味着在这些请求之后发出的每个请求都将被处理，直到它们完成。从文档中引用：

WSGI 是同步接口，而 Tornado 的并发模型是 基于单线程异步执行。这意味着 使用 Tornado 的 WSGIContainer 运行 WSGI 应用程序的可扩展性不如 在多线程 WSGI 服务器（如 gunicorn 或 uwsgi。

我已经尝试过：

1. 继续使用WSGIContainer，但在一个会使调用异步的处理程序中。没有成功。我得到：RuntimeError: There is no current event loop in thread 'ThreadPoolExecutor-0_0'

    class AsyncFallbackHandler(tornado.web.RequestHandler):
    def initialize(
        self, fallback: Callable[[httputil.HTTPServerRequest], None]
    ) -> None:
        self.fallback = fallback

    async def prepare(self, *args, **kwargs):
        await self.run_in_executor()
        self._finished = True
        self.on_finish()

    async def run_in_executor(self):
        loop = tornado.ioloop.IOLoop.instance().asyncio_loop
        done, pending = await asyncio.wait(
            fs=[loop.run_in_executor(None, self.fallback, self.request)],
            return_when=asyncio.ALL_COMPLETED
        )

1. 创建另一个不使用 WSGIContainer 的 RequestHandler。但是在这里，当请求是 404 时，它无法对 ConnexionResponse 进行 json 编码。我也不能将ConnexionResponse 写入管道，因为它必须是字符串/字节/字典。

class WifiRequestHandler(tornado.web.RequestHandler):
    async def get(self, *args, **kwargs):
        # await tornado.gen.sleep(20)
        ids = self.get_arguments('ids')
        method = get_wifi
        if self.request.path.startswith('/v1/wifi/all'):
            method = get_wifi_all

        self.write(await self.run_in_executor(method, ids))
        self.set_header('Content-Type', 'application/json')

    async def post(self, *args, **kwargs):
        logger.info(kwargs)
        body = tornado.escape.json_decode(self.request.body)
        self.write(await self.run_in_executor(update_wifi, body))
        self.set_header('Content-Type', 'application/json')

    async def run_in_executor(self, method, *args):
        loop = tornado.ioloop.IOLoop.instance().asyncio_loop
        done, pending = await asyncio.wait(
            fs=[loop.run_in_executor(None, method, args)],
            return_when=asyncio.ALL_COMPLETED
        )
        result = done.pop().result()
        if type(result) is ConnexionResponse:
            return result
        enc = JSONEncoder()
        return enc.encode(result)

请帮助我找到一种方法使我的一些端点异步

Answer 1

我通过解压缩 ConnexionResponse 并设置 status_code 来实现我的第二个解决方案。这是修复后的样子：

async def run_in_executor(self, method, *args):
    loop = tornado.ioloop.IOLoop.instance().asyncio_loop
    done, pending = await asyncio.wait(
        fs=[loop.run_in_executor(None, method, args)],
        return_when=asyncio.ALL_COMPLETED
    )
    result = done.pop().result()
    if type(result) is ConnexionResponse:
        self.set_status(result.status_code)
        return result.body
    enc = JSONEncoder()
    return enc.encode(result)