Postgres 和 scoped_session，但仍然致命：剩余的连接槽保留用于非复制超级用户连接答案

【问题标题】：Postgres and scoped_session, but still got FATAL: remaining connection slots are reserved for non-replication superuser connectionsPostgres 和 scoped_session，但仍然致命：剩余的连接槽保留用于非复制超级用户连接
【发布时间】：2020-09-22 13:01:30
【问题描述】：

我正在使用 sqlalchemy sessionmaker 和 scoped_session 为我的线程创建一个连接池，这样我就可以避免标题中的错误，但不幸的是我仍然得到它。我一直在阅读类似的questions 和blogs，但不幸的是我仍然无法解决这个问题。我的应用程序正在监听 pubsub，并在数据到达时在数据库中写入一些内容。该应用程序会收到大量消息，因此在达到一定数量后我会收到错误消息。我认为使用sessionmaker 和scoped_session 可以轻松处理这种情况，但我显然遗漏了一些东西。这是简化的代码：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from google.cloud import pubsub_v1

TOPIC_SUBSCRIBER = os.environ.get('PUBSUB_SUBSCRIBER')
PROJECT_ID = os.environ.get('PROJECT_ID')
client = pubsub_v1.SubscriberClient()
subscription_path = client.subscription_path(PROJECT_ID, TOPIC_SUBSCRIBER)

db_uri = os.environ.get('DATABASE_URI')
engine = create_engine(db_uri)

session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

def event_handler(message):

    session_db = Session()

    # Do stuff

    Session.remove()


def run():
    streaming_pull_future = client.subscribe(
        subscription_path, callback=event_handler
    )
    print("Listening for messages on {}".format(subscription_path))

    # Calling result() on StreamingPullFuture keeps the main thread from
    # exiting while messages get processed in the callbacks.
    try:
        streaming_pull_future.result()
    except Exception as e:  # noqa
        streaming_pull_future.cancel()
        print("ERROR: {}".format(str(e)))


if __name__ == '__main__':
    run()

【问题讨论】：

标签： python sqlalchemy google-cloud-pubsub

【解决方案1】：

该错误是由于 pubsub 创建了太多线程并使用太多连接使数据库过载。我通过限制可以同时处理的并发消息的数量来解决它。该方法也在here 中进行了讨论。这是我的代码：

def run():
    # Limit the subscriber to only have ten outstanding messages at a time.
    flow_control = pubsub_v1.types.FlowControl(max_messages=MAX_WORKERS)
    custom_executor = concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS)
    custom_scheduler = pubsub_v1.subscriber.scheduler.ThreadScheduler(custom_executor)

    streaming_pull_future = client.subscribe(
        subscription_path, callback=callback, flow_control=flow_control, scheduler=custom_scheduler
    )
    print("Listening for messages on {}".format(subscription_path))

    # Calling result() on StreamingPullFuture keeps the main thread from
    # exiting while messages get processed in the callbacks.
    try:
        streaming_pull_future.result()
    except Exception as e:  # noqa
        streaming_pull_future.cancel()
        print("ERROR: {}".format(str(e)))

上面的代码解决了这个问题，但是我认为sessionmaker 和scoped_session 会自己管理并发，而不是我必须在其他地方处理它。我很想听听一些 SQLAlchemy 专家对此的意见。我希望这对将来的人有所帮助。

【讨论】：