python - 共享 txpostgres 连接池

Question

我们有一个 RESTful(-ish) 扭曲的应用程序，它使用 txpostgres 访问 postgres 数据库。目前，每次客户端 ping 服务器以进行数据库调用时，我们都会生成新的 txpostgres.Connection 实例。这是低效的，导致我们的数据库很快不堪重负。我一直在尝试调整它以改用 txpostgres.ConnectionPool，但遇到了麻烦。现在我有一些看起来像这样的东西:

class DBTester(object):
    def __init__(self):
        self.cfg = load_config('local')  # load the db settings from a JSON file
        self.pool = ConnectionPool(None, min=1, **self.cfg) # create the pool

    @defer.inlineCallbacks
    def get_pool(self):
        yield self.pool.start()
        defer.returnValue(self.pool)


class DBT(object):
    def __init__(self):
        self.db = DBTester()

    @defer.inlineCallbacks
    def t(self):
        conn = yield self.db.get_pool()
        res = yield conn.runQuery('select * from clients')
        println('DBT.t result: {}'.format(res))


if __name__ == "__main__":
    dbt = DBT()
    dbt.t()
    dbt.t()

    reactor.run()

问题是 pool.start() 调用的时间。如果我把它放在 DBTester.__init__ 中，我会得到 psycopg2.OperationalError: asynchronous connection attempt underway。如果我把它放在 DBTester.get_pool 中，一个 db.t() 调用有效，而其他调用失败并显示 exceptions.AttributeError: 'NoneType'对象没有属性“runQuery”。我基本上整天都在为此苦苦挣扎，但一直无法破解，也无法在网上找到很多东西。

我真的只需要一个指向如何使用 ConnectionPool 的最小示例的指针。有什么建议吗？

Answer 1

听起来您的问题不在于 txpostgres，而在于扭曲和异步的思维方式。

exceptions.AttributeError: 'NoneType' object has no attribute 'runQuery' 表示:在建立连接之前，您试图在数据库之后抛出 SQL 查询。那是愚蠢的!所以现在我想我会抛出一个异常，以便亲爱的用户了解这种疯狂行为。

所以，如果你有类似的东西，这可能会发生

pool = ConnectionPool(None, min=1)
d1 = pool.start()
d2 = pool.runQuery('select tablename from pg_tables')

此代码在 react 器中创建了两个 deferred 和 thorws em。只有调度算法知道两者中哪一个先执行，如果是d2，则错误发生。

txpostgres.txpostgres.AlreadyConnected 表示: self 解释，启动一个已经启动的池是没有意义的。

psycopg2.OperationalError:正在进行异步连接尝试意味着:
当您开始执行 SQL 语句时，我正在建立一个很好的异步数据库连接。数据库连接还没有准备好，因此没有执行 sql 查询这让我很难过。我想我会抛出一个操作错误，这样亲爱的用户就知道语句失败了。

好的，所以我们需要一种方法来确保在我们向数据库发出 sql 查询之前建立连接。下面是一个使用回调来实现此目的的代码示例。

from txpostgres.txpostgres import ConnectionPool
from twisted.internet import reactor, defer
from twisted.python import log, util


class SomeClass(object):

    pool = ConnectionPool(
        None,
        min=1,
        user="user",
        password="pass",
        host='host.com')

    @defer.inlineCallbacks
    def fetch_tables(self):
        res = yield self.pool.runQuery('select tablename from pg_tables')
        defer.returnValue(res)


if __name__ == "__main__":

    def querydb(n=10):
        dl = []
        for i in range(n):
            d = s.fetch_tables()
            d.addCallback(lambda tables: util.println(len(tables)))
            dl.append(d)
        return defer.DeferredList(dl)

    s = SomeClass()
    d_startpool = s.pool.start()
    d_startpool.addCallback(lambda _: querydb())
    d_startpool.addCallback(lambda _: s.pool.close())
    d_startpool.addErrback(log.err)
    d_startpool.addBoth(lambda _: reactor.stop())

    reactor.run()

希望这对您有所帮助。