Не работает синглтон Pulsar Producer с помощью Python-клиента

У меня возникают трудности с созданием синглтонного продюсера Apache Pulsar в приложении Django. У меня есть Django-приложение, которое должно производить сотни сообщений каждую секунду. Вместо того чтобы каждый раз создавать нового производителя, я хочу повторно использовать того же производителя, который был создан в первый раз.

Проблема, с которой я столкнулся, заключается в том, что продюсер успешно создается, я могу видеть логи. но если я пытаюсь отправить сообщение с помощью продюсера, он застревает без какой-либо ошибки. Ниже приведен мой код для создания продюсера.

import logging
import os
import threading

import pulsar

from sastaticketpk.settings.config import PULSAR_ENABLED

logger = logging.getLogger(__name__)

PULSAR_ENV = os.environ.get("PULSAR_CONTAINER")

class PulsarClient:
    __producer = None
    __lock = threading.Lock()  # A lock to ensure thread-safe initialization of the singleton producer

    def __init__(self):
        """ Virtually private constructor. """
        raise RuntimeError("Call get_producer() instead")

    @classmethod
    def initialize_producer(cls):
        if PULSAR_ENABLED and PULSAR_ENV:
            logger.info("Creating pulsar producer")
            client = pulsar.Client(
                "pulsar://k8s-tooling-pulsarpr-7.elb.ap-southeast-1.amazonaws.com:6650"
            )
            # Create the producer
            try:
                cls.__producer = client.create_producer(
                    "persistent://public/default/gaf",
                    send_timeout_millis=1000
                )
                logger.info("Producer created successfully")
            except pulsar._pulsar.TopicNotFound as e:
                logger.error(f"Error creating producer: {e}")
            except Exception as e:
                logger.error(f"Error creating producer: {e}")

    @classmethod
    def get_producer(cls):
        logger.info(f"Producer value {cls.__producer}")
        if cls.__producer is None:
            with cls.__lock:
                if cls.__producer is None:
                    cls.initialize_producer()
        logger.info(f"Is producer connected {cls.__producer}")
        return cls.__producer

Я импортирую его в один из файлов утилиты, запускаемой при старте приложения. Ниже приведен код вызова:

PULSAR_PRODUCER = PulsarClient.get_producer()

Приведенный выше код отлично работает на моей локальной установке, но когда тот же код развернут на производстве (т.е. на AWS ECS), он не работает. Если я подключаюсь по SSH к контейнеру Docker, а затем запускаю оболочку Python и пытаюсь импортировать PULSAR_PRODUCER, он отображается как None. Пожалуйста, подскажите, что я могу упустить, так как я действительно застрял в понимании.

Вернуться на верх