Đảm bảo tính ổn định trong trình nghe thông báo PostgreSQL chạy lâu dài với Psycopg3

Đảm bảo thông báo đáng tin cậy trong trình nghe PostgreSQL chạy dài

Triển khai chương trình phụ trợ bằng Python và psycopg3 để xử lý các kết nối cơ sở dữ liệu dài hạn

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

Phương pháp tiếp cận thay thế: Sử dụng psycopg3 không đồng bộ để tăng cường khả năng phản hồi

Triển khai không đồng bộ bằng asyncio và psycopg3 của 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())

Kiểm thử đơn vị về độ bền

Kiểm tra đơn vị Python cho logic phụ trợ bằng cách sử dụng 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()

Câu hỏi thường gặp về Quản lý Trình nghe thông báo PostgreSQL

1. Mục đích của là gì conn.notifies() trong psycopg3?
2. conn.notifies() được sử dụng để truy xuất các thông báo được gửi bởi máy chủ PostgreSQL, cho phép xử lý sự kiện theo thời gian thực trong các ứng dụng.
3. Có thể LISTEN lệnh bị mất tin nhắn trong quá trình kết nối lại?
4. Không, PostgreSQL đệm các thông báo để tin nhắn không bị mất trong quá trình kết nối lại. Tuy nhiên, việc xử lý đúng đắn các notifies máy phát điện là cần thiết để đảm bảo xử lý liền mạch.
5. Tại sao tôi nên sử dụng autocommit=True?
6. Cài đặt autocommit=True cho phép kết nối áp dụng ngay các lệnh như LISTEN mà không cần chờ đợi một cam kết rõ ràng, cải thiện khả năng phản hồi.
7. Làm cách nào tôi có thể thực hiện kiểm tra tình trạng trong thời gian dài notifies quá trình?
8. Bạn có thể thực hiện định kỳ các truy vấn nhẹ như SELECT 1 để đảm bảo kết nối vẫn đáp ứng.
9. Các phương pháp hay nhất để dọn dẹp kết nối cơ sở dữ liệu là gì?
10. Sử dụng một with câu lệnh hoặc trình quản lý bối cảnh của Python đảm bảo rằng kết nối được đóng đúng cách, tránh rò rỉ tài nguyên.
11. Làm cách nào để xử lý các trường hợp ngoại lệ hết thời gian chờ trong conn.notifies()?
12. Bọc conn.notifies() trong khối thử ngoại trừ để nắm bắt các ngoại lệ hết thời gian chờ và xử lý chúng một cách khéo léo, chẳng hạn như bằng cách ghi nhật ký hoặc thử lại.
13. psycopg3 có hỗ trợ các hoạt động không đồng bộ cho thông báo không?
14. Có, psycopg3 cung cấp API không đồng bộ thông qua AsyncConnection, lý tưởng cho các ứng dụng không bị chặn, có thể mở rộng.
15. Điều gì xảy ra nếu tôi không đóng notifies máy phát điện?
16. Không đóng trình tạo có thể dẫn đến rò rỉ bộ nhớ hoặc treo tài nguyên, đặc biệt là trong các quy trình chạy dài.
17. Thông báo có thể bị bỏ qua trong một pg_sleep() hoạt động?
18. Có, thông báo được tạo trong thời gian ngủ có thể bị bỏ qua nếu không được lưu vào bộ nhớ đệm, đó là lý do tại sao việc xử lý đúng cách LISTEN lệnh là rất quan trọng.
19. Có an toàn khi sử dụng lại cùng một kết nối cho nhiều thông báo không?
20. Có, miễn là việc kiểm tra tình trạng và kết nối lại thích hợp được quản lý thì việc sử dụng lại cùng một kết nối sẽ hiệu quả và thân thiện với tài nguyên.
21. Làm cách nào để kiểm tra độ tin cậy của hệ thống thông báo của tôi?
22. Viết bài kiểm tra đơn vị bằng các thư viện như unittest.mock để mô phỏng các thông báo và hoạt động của cơ sở dữ liệu mà không cần dựa vào máy chủ trực tiếp.

Đảm bảo lắng nghe thông báo đáng tin cậy

Duy trì tình trạng kết nối cho các quy trình chạy lâu dài là điều cần thiết để hoạt động không bị gián đoạn. Với các công cụ của psycopg3 như conn.notify(), các nhà phát triển có thể triển khai các hệ thống thông báo mạnh mẽ. Kiểm tra tình trạng thường xuyên giúp tránh các kết nối không phản hồi. Ví dụ bao gồm giám sát hệ thống kiểm kê để cập nhật trực tiếp nhằm ngăn chặn tình trạng ngừng hoạt động.

Đóng và mở lại trình tạo thông báo, kết hợp với các lệnh SQL nhẹ, đảm bảo cả hiệu suất và độ tin cậy. Những kỹ thuật này áp dụng cho nhiều trường hợp sử dụng khác nhau, từ cập nhật hậu cần đến cảnh báo tài chính. Những chiến lược như vậy giúp bảo vệ các ứng dụng quan trọng khỏi thời gian ngừng hoạt động, đảm bảo trải nghiệm người dùng liền mạch. ⚡