关于Psycopg3异步连接notifies生成器的连接异常处理及健康检查时通知丢失的咨询

环境信息 #

• Psycopg版本：3.2.3
• PostgreSQL版本：PostgreSQL 14.13 (Homebrew) on x86_64-apple-darwin23.6.0, compiled by Apple clang version 16.0.0 (clang-1600.0.26.4), 64-bit

我维护一个可能连续运行数月的长时进程，它依赖conn.notifies()来监听数据库连接上的通知。不过官方文档里没有明确说明，当监听连接变得无响应或者损坏时，这个方法的预期行为是什么，所以我猜测可能需要自己实现健康检查逻辑。

首先想确认一个最核心的问题：**如果连接出现问题，conn.notifies()是否一定会抛出异常？**如果这个是确定的，那我后续的健康检查逻辑都不需要做了，有没有了解这块的朋友能给个准信？

假设conn.notifies()不一定会在连接异常时抛出异常，我看到有说法提到，可以在监听连接上执行语句而不会丢失通知——具体的操作步骤是：如果需要在该连接上执行新的LISTEN操作，可以先关闭notifies生成器（用generators.close()），执行完语句后再重新调用notifies()，消息会被libpq连接缓冲，所以不会丢失。

但我实际写了测试代码验证，发现结果和这个说法不符，以下是我的测试代码：

def _listen_for_notifications():
    with psycopg.Connection.connect("some_connection_string", autocommit=True) as conn:
        listen_sql = sql.SQL("LISTEN {}").format(sql.Identifier("some_channel_name"))

        conn.execute(listen_sql)
        gen = conn.notifies(timeout=5)
        print('listening')
        # 打印5秒超时窗口内收到的通知
        for n in gen:
            print(n)

        gen.close()  # 测试发现加不加这行，结果都一样

        print("Performing health check")
        # 模拟健康检查的耗时操作，实际中会用SELECT 1
        conn.execute("select pg_sleep(3)")

        conn.execute(listen_sql)
        gen = conn.notifies(timeout=5)
        print('listening again on the same connection')
        # 打印新一轮5秒超时窗口内收到的通知
        for n in gen:
            print(n)

        print('done!')


_listen_for_notifications()

测试场景与结果 #

我在三个不同的时间点发送了三条通知：

1. 第一个5秒的监听窗口内，发送payload为1的通知；
2. 在pg_sleep(3)执行的3秒期间，发送payload为2的通知；
3. 第二个5秒的监听窗口内，发送payload为3的通知。

实际运行输出如下：

listening
Notify(channel='some_channel_name', payload='1', pid=35237)
Performing health check
listening again on the same connection
Notify(channel='some_channel_name', payload='3', pid=35237)
done!

可以看到，payload为2的通知直接丢失了，并没有被libpq缓冲下来。这和我之前看到的说法不一致，想请教大家这是哪里出了问题？或者我的操作步骤有什么遗漏吗？

备注：内容来源于stack exchange，提问作者codeech