Säkerställa stabilitet i långvariga PostgreSQL-aviseringslyssnare med Psycopg3

Säkerställa tillförlitliga meddelanden i långvariga PostgreSQL-lyssnare

Backend-implementering med Python och psycopg3 för att hantera långvariga databasanslutningar

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

Alternativt tillvägagångssätt: Använd asynkron psycopg3 för förbättrad lyhördhet

Asynkron implementering med Pythons asyncio och 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())

Enhetstestning för robusthet

Python-enhetstester för backend-logik med hjälp av 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()

Vanliga frågor om hantering av PostgreSQL-aviseringslyssnare

1. Vad är syftet med conn.notifies() i psychopg3?
2. conn.notifies() används för att hämta meddelanden som skickats av PostgreSQL-servern, vilket möjliggör händelsehantering i realtid i applikationer.
3. Burk LISTEN kommandon förlorar meddelanden under återanslutningen?
4. Nej, PostgreSQL buffrar aviseringar så att meddelanden inte går förlorade under återanslutningen. Men korrekt hantering av notifies generator krävs för att säkerställa sömlös bearbetning.
5. Varför ska jag använda autocommit=True?
6. Miljö autocommit=True tillåter anslutningen att omedelbart tillämpa kommandon som LISTEN utan att vänta på ett uttryckligt åtagande, vilket förbättrar lyhördheten.
7. Hur kan jag utföra hälsokontroller under en långvarig notifies behandla?
8. Du kan med jämna mellanrum utföra lätta frågor som SELECT 1 för att säkerställa att anslutningen förblir responsiv.
9. Vilka är de bästa metoderna för att rensa upp databasanslutningar?
10. Att använda en with statement eller Pythons kontexthanterare säkerställer att anslutningen är ordentligt stängd, vilket undviker resursläckor.
11. Hur hanterar jag timeout-undantag i conn.notifies()?
12. Sjal conn.notifies() i ett försök-utom-block för att fånga undantag för timeout och hantera dem på ett elegant sätt, till exempel genom att logga eller försöka igen.
13. Stöder psycopg3 asynkrona operationer för aviseringar?
14. Ja, psycopg3 erbjuder ett asynkront API via AsyncConnection, vilket är idealiskt för icke-blockerande, skalbara applikationer.
15. Vad händer om jag inte stänger notifies generator?
16. Att misslyckas med att stänga generatorn kan resultera i minnesläckor eller hängande resurser, särskilt i långa processer.
17. Kan aviseringar missas under en pg_sleep() drift?
18. Ja, aviseringar som genereras under viloperioden kan missas om de inte buffras, vilket är anledningen till korrekt hantering av LISTEN kommandon är avgörande.
19. Är det säkert att återanvända samma anslutning för flera aviseringar?
20. Ja, så länge hälsokontroller och korrekta återanslutningar hanteras är återanvändning av samma anslutning effektivt och resursvänligt.
21. Hur kan jag testa tillförlitligheten hos mitt meddelandesystem?
22. Skriv enhetstester med hjälp av bibliotek som unittest.mock att simulera aviseringar och databasbeteende utan att förlita sig på en liveserver.

Säkerställer tillförlitlig avlyssning av meddelanden

Att upprätthålla anslutningshälsan för långvariga processer är avgörande för oavbruten drift. Med psycopg3s verktyg som conn.notifies(), kan utvecklare implementera robusta meddelandesystem. Regelbundna hälsokontroller hjälper till att undvika att anslutningar svarar. Exempel inkluderar övervakning av lagersystem för liveuppdateringar för att förhindra avbrott.

Stängning och återöppning av aviseringsgeneratorn, kombinerat med lätta SQL-kommandon, säkerställer både prestanda och tillförlitlighet. Dessa tekniker gäller för olika användningsfall, från logistikuppdateringar till ekonomiska varningar. Sådana strategier hjälper till att skydda kritiska applikationer mot stillestånd, vilket säkerställer en sömlös användarupplevelse. ⚡