Zapewnienie stabilności długotrwałych odbiorników powiadomień PostgreSQL za pomocą Psycopg3

Zapewnienie niezawodnych powiadomień w długotrwałych odbiornikach PostgreSQL

Implementacja backendu przy użyciu języka Python i psycopg3 do obsługi długotrwałych połączeń z bazami danych

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

Podejście alternatywne: użycie asynchronicznego psycopg3 w celu zwiększenia responsywności

Implementacja asynchroniczna przy użyciu asyncio i psycopg3 Pythona

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

Testowanie jednostkowe pod kątem wytrzymałości

Testy jednostkowe Pythona dla logiki zaplecza przy użyciu testu jednostkowego

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

Często zadawane pytania dotyczące zarządzania odbiornikami powiadomień PostgreSQL

1. Jaki jest cel conn.notifies() w psycopg3?
2. conn.notifies() służy do pobierania powiadomień wysyłanych przez serwer PostgreSQL, umożliwiając obsługę zdarzeń w aplikacjach w czasie rzeczywistym.
3. Móc LISTEN polecenia tracą wiadomości podczas ponownego połączenia?
4. Nie, PostgreSQL buforuje powiadomienia, więc wiadomości nie zostaną utracone podczas ponownego połączenia. Jednak właściwe obchodzenie się z notifies Aby zapewnić płynne przetwarzanie, wymagany jest generator.
5. Dlaczego powinienem używać autocommit=True?
6. Ustawienie autocommit=True umożliwia połączeniu natychmiastowe zastosowanie poleceń takich jak LISTEN bez czekania na wyraźne zatwierdzenie, co poprawia responsywność.
7. Jak mogę przeprowadzić kontrolę stanu zdrowia podczas długotrwałego notifies proces?
8. Możesz okresowo wykonywać lekkie zapytania, takie jak SELECT 1 aby mieć pewność, że połączenie będzie responsywne.
9. Jakie są najlepsze praktyki czyszczenia połączeń z bazami danych?
10. Korzystanie z with instrukcja lub menedżer kontekstu Pythona zapewnia prawidłowe zamknięcie połączenia, unikając wycieków zasobów.
11. Jak obsługiwać wyjątki dotyczące limitu czasu w conn.notifies()?
12. Zawinąć conn.notifies() w bloku try-except, aby wychwycić wyjątki związane z przekroczeniem limitu czasu i sprawnie je obsłużyć, na przykład poprzez zalogowanie się lub ponowną próbę.
13. Czy psycopg3 obsługuje operacje asynchroniczne dla powiadomień?
14. Tak, psycopg3 oferuje asynchroniczne API poprzez AsyncConnection, który idealnie nadaje się do nieblokujących, skalowalnych aplikacji.
15. Co się stanie, jeśli nie zamknę notifies generator?
16. Niezamknięcie generatora może spowodować wycieki pamięci lub zawieszenie zasobów, szczególnie w długotrwałych procesach.
17. Czy powiadomienia mogą zostać pominięte podczas pg_sleep() działanie?
18. Tak, powiadomienia generowane w okresie uśpienia mogą zostać pominięte, jeśli nie są buforowane, dlatego należy je odpowiednio obsłużyć LISTEN polecenia są kluczowe.
19. Czy ponowne użycie tego samego połączenia do wielu powiadomień jest bezpieczne?
20. Tak, jeśli zarządzane są kontrole stanu i prawidłowe ponowne połączenia, ponowne użycie tego samego połączenia jest wydajne i oszczędza zasoby.
21. Jak mogę przetestować niezawodność mojego systemu powiadomień?
22. Pisz testy jednostkowe, korzystając z bibliotek takich jak unittest.mock do symulacji powiadomień i zachowania bazy danych bez polegania na działającym serwerze.

Zapewnienie niezawodnego słuchania powiadomień

Utrzymywanie dobrej kondycji połączenia w przypadku długotrwałych procesów jest niezbędne dla nieprzerwanego działania. Dzięki narzędziom psycopg3, takim jak połącz.powiadamia()programiści mogą wdrożyć niezawodne systemy powiadomień. Regularne kontrole stanu pomagają uniknąć nieodpowiadających połączeń. Przykłady obejmują monitorowanie systemów zapasów pod kątem aktualizacji na żywo, aby zapobiec przestojom.

Zamykanie i ponowne otwieranie generatora powiadomień w połączeniu z lekkimi poleceniami SQL zapewnia zarówno wydajność, jak i niezawodność. Techniki te mają zastosowanie w różnych przypadkach użycia, od aktualizacji logistycznych po alerty finansowe. Takie strategie pomagają chronić krytyczne aplikacje przed przestojami, zapewniając bezproblemową obsługę użytkownika. ⚡