Memastikan Kestabilan dalam Pendengar Pemberitahuan PostgreSQL Jangka Panjang dengan Psycopg3

Memastikan Pemberitahuan Boleh Dipercayai dalam Pendengar PostgreSQL yang Berlangsung Lama

Pelaksanaan backend menggunakan Python dan psycopg3 untuk mengendalikan sambungan pangkalan data yang berjalan lama

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()

Pendekatan Alternatif: Menggunakan Psycopg3 Asynchronous untuk Responsif Ditingkatkan

Pelaksanaan tak segerak menggunakan asyncio dan psycopg3 Python

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())

Ujian Unit untuk Kekukuhan

Ujian unit Python untuk logik belakang menggunakan unittest

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()

Soalan Lazim tentang Mengurus Pendengar Pemberitahuan PostgreSQL

1. Apakah tujuan conn.notifies() dalam psycopg3?
2. conn.notifies() digunakan untuk mendapatkan semula pemberitahuan yang dihantar oleh pelayan PostgreSQL, membolehkan pengendalian acara masa nyata dalam aplikasi.
3. boleh LISTEN arahan kehilangan mesej semasa penyambungan semula?
4. Tidak, PostgreSQL menampan pemberitahuan, jadi mesej tidak hilang semasa penyambungan semula. Walau bagaimanapun, pengendalian yang betul notifies penjana diperlukan untuk memastikan pemprosesan yang lancar.
5. Mengapa saya perlu menggunakan autocommit=True?
6. Tetapan autocommit=True membenarkan sambungan untuk segera menggunakan arahan seperti LISTEN tanpa menunggu komitmen yang jelas, meningkatkan tindak balas.
7. Bagaimanakah saya boleh melakukan pemeriksaan kesihatan semasa jangka masa panjang notifies proses?
8. Anda boleh melaksanakan pertanyaan ringan seperti secara berkala SELECT 1 untuk memastikan sambungan kekal responsif.
9. Apakah amalan terbaik untuk membersihkan sambungan pangkalan data?
10. Menggunakan a with pernyataan atau pengurus konteks Python memastikan sambungan ditutup dengan betul, mengelakkan kebocoran sumber.
11. Bagaimanakah saya mengendalikan pengecualian tamat masa dalam conn.notifies()?
12. Bungkus conn.notifies() dalam blok cuba-kecuali untuk menangkap pengecualian tamat masa dan mengendalikannya dengan anggun, seperti dengan mengelog atau mencuba semula.
13. Adakah psycopg3 menyokong operasi tak segerak untuk pemberitahuan?
14. Ya, psycopg3 menawarkan API tak segerak melalui AsyncConnection, yang sesuai untuk aplikasi tidak menyekat dan berskala.
15. Apa yang berlaku jika saya tidak menutup notifies penjana?
16. Gagal menutup penjana boleh mengakibatkan kebocoran memori atau sumber yang tergantung, terutamanya dalam proses yang berjalan lama.
17. Bolehkah pemberitahuan terlepas semasa a pg_sleep() operasi?
18. Ya, pemberitahuan yang dijana semasa tempoh tidur mungkin terlepas jika tidak ditimbal, itulah sebabnya pengendalian yang betul LISTEN perintah adalah penting.
19. Adakah selamat untuk menggunakan semula sambungan yang sama untuk berbilang pemberitahuan?
20. Ya, selagi pemeriksaan kesihatan dan penyambungan semula yang betul diuruskan, penggunaan semula sambungan yang sama adalah cekap dan mesra sumber.
21. Bagaimanakah saya boleh menguji kebolehpercayaan sistem pemberitahuan saya?
22. Tulis ujian unit menggunakan perpustakaan seperti unittest.mock untuk mensimulasikan pemberitahuan dan gelagat pangkalan data tanpa bergantung pada pelayan langsung.

Memastikan Pendengaran Pemberitahuan Boleh Dipercayai

Mengekalkan kesihatan sambungan untuk proses yang berjalan lama adalah penting untuk operasi tanpa gangguan. Dengan alat psycopg3 seperti conn.notifies(), pembangun boleh melaksanakan sistem pemberitahuan yang mantap. Pemeriksaan kesihatan yang kerap membantu mengelakkan sambungan tidak responsif. Contohnya termasuk memantau sistem inventori untuk kemas kini langsung bagi mengelakkan gangguan.

Menutup dan membuka semula penjana pemberitahuan, digabungkan dengan arahan SQL yang ringan, memastikan prestasi dan kebolehpercayaan. Teknik ini digunakan untuk pelbagai kes penggunaan, daripada kemas kini logistik kepada makluman kewangan. Strategi sedemikian membantu melindungi aplikasi kritikal daripada masa henti, memastikan pengalaman pengguna yang lancar. ⚡