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-Exception 블록에서 사용합니다.
13. psycopg3는 알림에 대한 비동기 작업을 지원합니까?
14. 예, psycopg3는 다음을 통해 비동기 API를 제공합니다. AsyncConnection이는 비차단 확장형 애플리케이션에 이상적입니다.
15. 닫지 않으면 어떻게 되나요? notifies 발전기?
16. 생성기를 닫지 못하면 특히 장기 실행 프로세스에서 메모리 누수나 리소스 중단이 발생할 수 있습니다.
17. 알림 중에 알림이 누락될 수 있나요? pg_sleep() 작업?
18. 예, 버퍼링되지 않으면 절전 기간 동안 생성된 알림이 누락될 수 있습니다. LISTEN 명령이 중요합니다.
19. 여러 알림에 동일한 연결을 재사용해도 안전합니까?
20. 예, 상태 확인과 적절한 재연결이 관리된다면 동일한 연결을 재사용하는 것이 효율적이고 리소스 친화적입니다.
21. 알림 시스템의 신뢰성을 어떻게 테스트할 수 있나요?
22. 다음과 같은 라이브러리를 사용하여 단위 테스트 작성 unittest.mock 라이브 서버에 의존하지 않고 알림 및 데이터베이스 동작을 시뮬레이션합니다.

안정적인 알림 청취 보장

중단 없는 작업을 위해서는 장기 실행 프로세스의 연결 상태를 유지하는 것이 필수적입니다. psycopg3의 도구를 사용하면 다음과 같습니다. conn.notify(), 개발자는 강력한 알림 시스템을 구현할 수 있습니다. 정기적인 상태 확인을 통해 응답하지 않는 연결을 방지할 수 있습니다. 예를 들어, 중단을 방지하기 위해 실시간 업데이트를 위한 인벤토리 시스템 모니터링이 포함됩니다.

경량 SQL 명령과 결합된 알림 생성기를 닫았다가 다시 열면 성능과 안정성이 모두 보장됩니다. 이러한 기술은 물류 업데이트부터 금융 알림까지 다양한 사용 사례에 적용됩니다. 이러한 전략은 다운타임으로부터 중요한 애플리케이션을 보호하고 원활한 사용자 경험을 보장하는 데 도움이 됩니다. ⚡