python Tornado 处理程序 IO 阻塞整个服务器网络

Question

我正在用 tornado 实现 REST api，并希望它是非阻塞的。

目前，与问题相关的代码是这样的:

class ReprsHandler(web.RequestHandler):
    async def get(self, name):

        db = await dbf.create_handler()

        if 'id' in list(self.request.query_arguments.keys()):
            db_future = asyncio.ensure_future(db.get_repr(name, self.get_query_argument('id')))
        else:
            db_future = asyncio.ensure_future(db.get_reprs(name))

        result = await db_future
        response = result.toSerializedStream()

        self.set_status(HTTPStatus.OK)
        self.write(response)
        self.set_header('Content-Type', 'text/plain')
        self.finish()


class App(object):
    def __init__(self, loop):
        self.server_app = web.Application(
            handlers=[
                (r"/api/v1/([a-zA-Z0-9_-]+)/reprs", ReprsHandler),
            ]
        )

def main():
    AsyncIOMainLoop().install()
    loop = asyncio.get_event_loop()
    app = App(loop)
    server = tornado.httpserver.HTTPServer(app.server_app, max_body_size=config['max_upload_size'], max_buffer_size=config['max_upload_size'])
    server.bind(config['server_port'])
    server.start()

    loop.run_forever()

简单的代码，但是数据比较大，所以发送完大概需要3~4分钟。

我希望处理程序的逻辑和网络 IO 都是非阻塞的，但它在发送数据作为响应时阻塞了服务器网络。逻辑很好。它们不会阻止其他请求。

详细信息:

• 此代码运行在 docker，ubuntu 16.04，使用 python 3.5 实现。
• 服务器使用 nginx 作为端口代理。

可能是什么问题？我不知道是什么导致了这个问题。

Answer 1

因为您提到 result.toSerializedStream() 会腌制数据。所以，是的，你是对的，阻塞是因为网络 io。

为避免这种情况，您可以在 block 中发送数据并在每次self.write() 之后调用self.flush()。调用 flush 会将响应写入网络。由于您可以在flush 上await，在数据写入网络套接字之前，协程将暂停，服务器不会阻塞。这允许其他处理程序异步运行。

代码示例:

async def get(self, name):
    ...
    response = result.toSerializedStream()

    chunk_size = 1024 * 1024 * 10 # 10 MiB

    start_byte = 0
    while True:
        chunk = response[start_byte : start_byte + chunk_size]
        if not chunk:
            break
        self.write(chunk)
        await self.flush() # wait while data is flushed to network

        start_byte += chunk_size # move start_byte forward

---

重要:

这里要注意的一件重要事情是 self.flush() 非常快。如果您将小数据刷新到网络，await 延迟非常小，协程会不停地运行，从而阻塞服务器。

在上面的示例代码中，我将 chunk_size 设置为 10 MiB，但是如果您的计算机速度很快，await 延迟将非常非常小，循环可能会不停地运行，直到发送完整个数据。

我鼓励您根据需要增加或减少 chunk_size 值。

---

进一步改进建议:

所有数据都在内存中。现在您的处理程序是异步的并且不会阻塞，如果在 ReprsHandler 处收到另一个请求，这将导致在内存中存储更多数据。如果收到越来越多的请求，那么您可以判断会发生什么。

为避免这种情况，您可以将其转储到文件中，而不是对内存中的数据进行酸洗。然后在您的处理程序中，只需打开该文件并分块读取并发送。