Tornado 中具有多个进程的异步连接池

Question

我正在使用 Tornado 4.2.1 和 tornadoes 2.4.1 库来查询我的 Elasticsearch 数据库，我正在寻找一种方法来初始化 连接池在多进程服务中的多个 RequestHandler 实例之间共享。

这有可能吗？是否有特定的库可以让 Tornado 做到这一点？

提前致谢

Answer 1

由于 tornado-es 只是一个 HTTP 客户端，它使用 AsyncHTTPClient在ESConnection . 每次请求都会建立新的 TCP 连接，除非指定了 Connection: keep-alive header 。

conn = ESConnection()
conn.httprequest_kwargs['headers'] = {'Connection': 'keep-alive'}

我没有测试过，但应该可以。我在 ruby​​ 中使用了类似的设置(使用 patron http 客户端)，并且运行良好

下一步

AsyncHTTPClient 对每个 ioloop 的最大同时请求数 (fetch) 有限制。每个达到限制的请求都只是在内部排队。

您可能想要增加全局限制:

AsyncHTTPClient.configure(None, max_clients=50)

或者用自己的限制分离客户端(force_instance):

from tornadoes import ESConnection
from tornado.httpclient import AsyncHTTPClient

class CustomESConnection(ESConnection):

    def __init__(self, , host='localhost', port='9200', io_loop=None, protocol='http', max_clients=20):
        super(CustomESConnection, self).__init__(host, port, io_loop, protocol)
        self.client = AsyncHTTPClient(force_instance=True, max_clients=max_clients)

最后

要重用相同的 ESConnection，您可以在应用程序中创建它，因为应用程序可用于每个请求 (RequestHandler)

from tornado.web import Application, RequestHandler
from tornadoes import ESConnection

class MainHandler(RequestHandler):
    def get(self):
        yield self.application.es.search('something')


class MyApp(Application):

    def __init__(self, *args, **kwargs):
        super(MyApp, self).__init__(*args, **kwargs)

        self.es = ESconnection()

if __name__ == "__main__":
    application = MyApp([
        (r"/", MainHandler),
    ])
    application.listen(8888)
    tornado.ioloop.IOLoop.current().start()

多进程

其实没有简单的方法。常见的方法是 pooler，它主要用于需要持久连接的情况，例如数据库(pgbouncer 对于 postgres)或作为高负载服务的优化。

而且你将不得不编写一个 pooler，一个到 es 的网关应用程序

subprocess1 
           \  (http, zmq, ...)
            \            
              > pooler (some queue and tornadoes api) - http -> elastisearch
            /
           /
subprocess2

子进程可以通过 HTTP 与 pooler 通信，ØMQ (甚至有很多示例 pooler)或 IPC 的一些实现( socks ，...)。