使用 Psycopg3 确保长时间运行的 PostgreSQL 通知监听器的稳定性

确保长时间运行的 PostgreSQL 侦听器中的可靠通知

使用Python和psycopg3的后端实现来处理长时间运行的数据库连接

import psycopg
from psycopg import sql
import time
CONN_STR = "postgresql://user:password@localhost/dbname"
def listen_notifications():
    try:
        with psycopg.connect(CONN_STR, autocommit=True) as conn:
            listen_sql = sql.SQL("LISTEN {};").format(sql.Identifier("scheduler_test"))
            conn.execute(listen_sql)
            print("Listening for notifications...")
            gen = conn.notifies(timeout=5)
            for notification in gen:
                print("Received notification:", notification)
                perform_health_check(conn, listen_sql)
    except Exception as e:
        print("Error:", e)
def perform_health_check(conn, listen_sql):
    try:
        print("Performing health check...")
        conn.execute("SELECT 1")
        conn.execute(listen_sql)
    except Exception as e:
        print("Health check failed:", e)
if __name__ == "__main__":
    listen_notifications()

替代方法：使用异步 psycopg3 增强响应能力

使用Python的asyncio和psycopg3进行异步实现

import asyncio
from psycopg import AsyncConnection, sql
CONN_STR = "postgresql://user:password@localhost/dbname"
async def listen_notifications():
    try:
        async with AsyncConnection.connect(CONN_STR, autocommit=True) as conn:
            listen_sql = sql.SQL("LISTEN {};").format(sql.Identifier("scheduler_test"))
            await conn.execute(listen_sql)
            print("Listening for notifications...")
            gen = conn.notifies(timeout=5)
            async for notification in gen:
                print("Received notification:", notification)
                await perform_health_check(conn, listen_sql)
    except Exception as e:
        print("Error:", e)
async def perform_health_check(conn, listen_sql):
    try:
        print("Performing health check...")
        await conn.execute("SELECT 1")
        await conn.execute(listen_sql)
    except Exception as e:
        print("Health check failed:", e)
if __name__ == "__main__":
    asyncio.run(listen_notifications())

稳健性单元测试

使用unittest对后端逻辑进行Python单元测试

import unittest
from unittest.mock import patch, MagicMock
class TestNotificationListener(unittest.TestCase):
    @patch("psycopg.connect")
    def test_listen_notifications(self, mock_connect):
        mock_conn = MagicMock()
        mock_connect.return_value.__enter__.return_value = mock_conn
        mock_conn.notifies.return_value = iter(["test_notification"])
        listen_notifications()
        mock_conn.execute.assert_called_with("LISTEN scheduler_test;")
        mock_conn.notifies.assert_called_once()
if __name__ == "__main__":
    unittest.main()

有关管理 PostgreSQL 通知监听器的常见问题解答

1. 目的是什么 conn.notifies() 在 psycopg3 中？
2. conn.notifies() 用于检索 PostgreSQL 服务器发送的通知，从而在应用程序中启用实时事件处理。
3. 能 LISTEN 命令在重新连接期间丢失消息？
4. 不会，PostgreSQL 会缓冲通知，因此在重新连接期间消息不会丢失。然而，正确处理 notifies 需要生成器来确保无缝处理。
5. 我为什么要使用 autocommit=True？
6. 环境 autocommit=True 允许连接立即应用命令，例如 LISTEN 无需等待显式提交，从而提高响应能力。
7. 如何在长时间运行期间进行健康检查 notifies 过程？
8. 您可以定期执行轻量级查询，例如 SELECT 1 以确保连接保持响应。
9. 清理数据库连接的最佳实践是什么？
10. 使用 with 语句或Python的上下文管理器确保连接正确关闭，避免资源泄漏。
11. 如何处理超时异常 conn.notifies()？
12. 裹 conn.notifies() 在 try-except 块中捕获超时异常并妥善处理它们，例如通过记录或重试。
13. psycopg3 支持通知的异步操作吗？
14. 是的，psycopg3 通过以下方式提供异步 API AsyncConnection，这是非阻塞、可扩展应用程序的理想选择。
15. 如果我不关闭会发生什么 notifies 发电机？
16. 未能关闭生成器可能会导致内存泄漏或挂起资源，尤其是在长时间运行的进程中。
17. 期间会错过通知吗 pg_sleep() 手术？
18. 是的，如果不缓冲，在睡眠期间生成的通知可能会丢失，这就是正确处理 LISTEN 命令至关重要。
19. 为多个通知重用同一个连接是否安全？
20. 是的，只要管理运行状况检查和正确的重新连接，重复使用相同的连接就是高效且资源友好的。
21. 如何测试通知系统的可靠性？
22. 使用类似的库编写单元测试 unittest.mock 无需依赖实时服务器即可模拟通知和数据库行为。

确保可靠的通知监听

保持长时间运行的进程的连接运行状况对于不间断的操作至关重要。使用 psycopg3 的工具，例如 conn.notify()，开发人员可以实现强大的通知系统。定期运行状况检查有助于避免连接无响应。示例包括监控库存系统进行实时更新以防止中断。

关闭并重新打开通知生成器，结合轻量级 SQL 命令，可确保性能和可靠性。这些技术适用于各种用例，从物流更新到财务警报。此类策略有助于保护关键应用程序免遭停机，确保无缝的用户体验。 ⚡