ضمان الاستقرار في مستمعي إشعارات PostgreSQL طويلة الأمد باستخدام Psycopg3

ضمان إشعارات موثوقة في مستمعي 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's 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

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

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 بيان أو مدير سياق بايثون يضمن إغلاق الاتصال بشكل صحيح، وتجنب تسرب الموارد.
11. كيف أتعامل مع استثناءات المهلة في conn.notifies()؟
12. طَوّق conn.notifies() في كتلة محاولة باستثناء التقاط استثناءات المهلة والتعامل معها بأمان، مثل تسجيل الدخول أو إعادة المحاولة.
13. هل يدعم psycopg3 العمليات غير المتزامنة للإشعارات؟
14. نعم، يقدم psycopg3 واجهة برمجة تطبيقات غير متزامنة عبر AsyncConnection، وهو مثالي للتطبيقات غير المحظورة والقابلة للتطوير.
15. ماذا يحدث إذا لم أغلق notifies مولد؟
16. قد يؤدي الفشل في إغلاق المولد إلى تسرب الذاكرة أو تعليق الموارد، خاصة في العمليات طويلة الأمد.
17. هل يمكن تفويت الإشعارات أثناء أ pg_sleep() عملية؟
18. نعم، قد يتم تفويت الإشعارات التي تم إنشاؤها أثناء فترة النوم إذا لم يتم تخزينها مؤقتًا، ولهذا السبب يتم التعامل معها بشكل صحيح LISTEN الأوامر أمر بالغ الأهمية.
19. هل من الآمن إعادة استخدام نفس الاتصال لإشعارات متعددة؟
20. نعم، طالما تتم إدارة عمليات التحقق من السلامة وعمليات إعادة الاتصال المناسبة، فإن إعادة استخدام نفس الاتصال تكون فعالة وصديقة للموارد.
21. كيف يمكنني اختبار مدى موثوقية نظام الإشعارات الخاص بي؟
22. اكتب اختبارات الوحدة باستخدام مكتبات مثل unittest.mock لمحاكاة الإخطارات وسلوك قاعدة البيانات دون الاعتماد على خادم مباشر.

ضمان الاستماع الموثوق للإشعارات

يعد الحفاظ على صحة الاتصال للعمليات طويلة الأمد أمرًا ضروريًا للعمليات غير المنقطعة. باستخدام أدوات psycopg3 مثل يخطر ()يمكن للمطورين تنفيذ أنظمة إعلام قوية. تساعد فحوصات السلامة المنتظمة على تجنب الاتصالات غير المستجيبة. تشمل الأمثلة مراقبة أنظمة المخزون للحصول على التحديثات المباشرة لمنع انقطاع الخدمة.

يؤدي إغلاق منشئ الإشعارات وإعادة فتحه، بالإضافة إلى أوامر SQL خفيفة الوزن، إلى ضمان الأداء والموثوقية. تنطبق هذه التقنيات على حالات الاستخدام المختلفة، بدءًا من التحديثات اللوجستية وحتى التنبيهات المالية. تساعد مثل هذه الاستراتيجيات على حماية التطبيقات الهامة من التوقف عن العمل، مما يضمن تجربة مستخدم سلسة. ⚡