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 通知リスナーの管理に関する FAQ

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-excel ブロッ​​ク内でタイムアウト例外をキャッチし、ログ記録や再試行などによって適切に処理します。
13. psycopg3 は通知の非同期操作をサポートしていますか?
14. はい、psycopg3 は非同期 API を提供します。 AsyncConnectionこれは、ノンブロッキングでスケーラブルなアプリケーションに最適です。
15. 閉じないとどうなりますか notifies ジェネレータ？
16. ジェネレーターを閉じないと、特に長時間実行プロセスの場合、メモリ リークやリソースのハングが発生する可能性があります。
17. 通知は途中で見逃される可能性があります pg_sleep() 手術？
18. はい、スリープ期間中に生成された通知はバッファリングされないと見逃される可能性があります。そのため、 LISTEN コマンドが重要です。
19. 複数の通知に同じ接続を再利用しても安全ですか?
20. はい、ヘルスチェックと適切な再接続が管理されている限り、同じ接続の再利用は効率的でリソースに優しいものになります。
21. 通知システムの信頼性をテストするにはどうすればよいですか?
22. 次のようなライブラリを使用して単体テストを作成します。 unittest.mock ライブサーバーに依存せずに通知とデータベースの動作をシミュレートします。

信頼性の高い通知リスニングの確保

長時間実行プロセスの接続の健全性を維持することは、中断のない操作にとって不可欠です。 psycopg3 のツールを使用すると、 conn.notify()、開発者は堅牢な通知システムを実装できます。定期的なヘルスチェックは、応答しない接続を回避するのに役立ちます。例としては、機能停止を防ぐために在庫システムを監視してライブ更新を行うことが挙げられます。

通知ジェネレーターを閉じて再度開くと、軽量の SQL コマンドと組み合わせることで、パフォーマンスと信頼性の両方が保証されます。これらの手法は、物流の最新情報から財務アラートまで、さまざまなユースケースに適用されます。このような戦略は、重要なアプリケーションをダウンタイムから保護し、シームレスなユーザー エクスペリエンスを保証するのに役立ちます。 ⚡