Ilgai veikiančių PostgreSQL pranešimų klausytojų stabilumo užtikrinimas naudojant Psycopg3

Patikimų pranešimų užtikrinimas ilgalaikiuose PostgreSQL klausytojuose

Backend įgyvendinimas naudojant Python ir psycopg3, kad būtų galima apdoroti ilgalaikius duomenų bazių ryšius

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

Alternatyvus metodas: asinchroninio psycopg3 naudojimas geresniam reagavimui

Asinchroninis įgyvendinimas naudojant Python asyncio ir 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())

Vieneto tvirtumo bandymas

„Python“ vienetų testavimas užpakalinei logikai naudojant 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()

DUK apie PostgreSQL pranešimų klausytojų valdymą

1. Koks tikslas conn.notifies() psycopg3?
2. conn.notifies() naudojamas PostgreSQL serverio siunčiamiems pranešimams gauti, kad programose būtų galima tvarkyti įvykius realiuoju laiku.
3. Gali LISTEN komandos praranda pranešimus prisijungiant iš naujo?
4. Ne, PostgreSQL saugo pranešimus, todėl pranešimai neprarandami prisijungiant iš naujo. Tačiau tinkamai elgiantis su notifies generatorius reikalingas, kad būtų užtikrintas sklandus apdorojimas.
5. Kodėl turėčiau naudoti autocommit=True?
6. Nustatymas autocommit=True leidžia ryšiui nedelsiant taikyti tokias komandas kaip LISTEN nelaukiant aiškaus įsipareigojimo, gerinant reagavimą.
7. Kaip aš galiu pasitikrinti sveikatą ilgą laiką notifies procesas?
8. Galite periodiškai vykdyti lengvas užklausas, pvz SELECT 1 kad ryšys išliktų reaguojantis.
9. Kokia yra geriausia duomenų bazių ryšių valymo praktika?
10. Naudojant a with pareiškimas arba Python konteksto tvarkyklė užtikrina, kad ryšys būtų tinkamai uždarytas, išvengiant išteklių nutekėjimo.
11. Kaip tvarkyti skirtojo laiko išimtis conn.notifies()?
12. Apvyniokite conn.notifies() „Try-Except“ bloke, kad gautumėte skirtojo laiko išimtis ir su jais elgtųsi maloniai, pavyzdžiui, registruodami arba bandydami dar kartą.
13. Ar psycopg3 palaiko asinchronines pranešimų operacijas?
14. Taip, psycopg3 siūlo asinchroninę API per AsyncConnection, kuris idealiai tinka neblokuojančioms, keičiamo dydžio programoms.
15. Kas atsitiks, jei neuždarysiu notifies generatorius?
16. Nepavykus uždaryti generatoriaus, gali nutekėti atmintis arba užstrigti ištekliai, ypač jei procesai vyksta ilgai.
17. Ar gali būti praleisti pranešimai per a pg_sleep() operacija?
18. Taip, per miego laikotarpį generuoti pranešimai gali būti praleisti, jei jie nebus saugomi, todėl tinkamai elgiamasi LISTEN komandos yra labai svarbios.
19. Ar saugu pakartotinai naudoti tą patį ryšį keliems pranešimams?
20. Taip, tol, kol atliekami sveikatos patikrinimai ir tinkami pakartotiniai prisijungimai, pakartotinis to paties ryšio naudojimas yra efektyvus ir nereikalauja išteklių.
21. Kaip galiu patikrinti savo pranešimų sistemos patikimumą?
22. Rašykite vienetų testus naudodami tokias bibliotekas kaip unittest.mock Norėdami imituoti pranešimus ir duomenų bazės elgesį, nepasitikėdami tiesioginiu serveriu.

Patikimo pranešimų klausymosi užtikrinimas

Norint užtikrinti nenutrūkstamą veikimą, labai svarbu palaikyti ryšio būklę ilgai vykstantiems procesams. Su psycopg3 įrankiais, pvz conn.notifyes(), kūrėjai gali įdiegti patikimas pranešimų sistemas. Reguliarūs sveikatos patikrinimai padeda išvengti nereaguojančių ryšių. Pavyzdžiai apima atsargų sistemų stebėjimą, kad būtų atnaujinami tiesioginiai naujiniai, siekiant išvengti gedimų.

Pranešimų generatoriaus uždarymas ir pakartotinis atidarymas kartu su lengvomis SQL komandomis užtikrina našumą ir patikimumą. Šie metodai taikomi įvairiems naudojimo atvejams – nuo ​​logistikos atnaujinimų iki finansinių įspėjimų. Tokios strategijos padeda apsaugoti svarbias programas nuo prastovų ir užtikrina sklandžią vartotojo patirtį. ⚡