你好,我正在尝试从Django应用程序连接到电子邮件队列。我已经创建了一个EmailPublisher,它将是一个扩展RabbitMQProducer对象的单例对象,该对象具有连接到rabbitmq服务的实际代码。现在我想分享gunicorn工作者之间的rabbitmq连接,我该怎么做?我用Python 3.7和Django。
class RabbitMQProducer(ABC):
def __init__(self):
self.credentials = pika.PlainCredentials(
settings.RABBITMQCONFIG['username'],
settings.RABBITMQCONFIG['password'],
)
self.parameters = pika.ConnectionParameters(
host=settings.RABBITMQCONFIG['host'],
port=settings.RABBITMQCONFIG['port'],
virtual_host=settings.RABBITMQCONFIG['virtual_host'],
credentials=self.credentials,
heartbeat=600,
blocked_connection_timeout=300,
client_properties={
'connection_name': self.get_connection_name(),
}
)
self.connection = None
self.channels = {}
self.connect()
def connect(self):
if not self.connection or self.connection.is_closed:
self.connection = BlockingConnection(self.parameters)
self.close_channels()
self.declare_channels()
def declare_channels(self):
for i in range(self.get_channel_count()):
self.channels[i] = self.assign_channel()
def send(self, message):
try:
self.connect()
self.thread_safe_publish(message)
except Exception as e:
Log.e(f"Failed to send message to RabbitMQ: {e}")
def assign_channel(self):
if not self.connection or self.connection.is_closed:
self.connect()
return None
channel = self.connection.channel(channel_number=None)
channel.exchange_declare(
exchange=self.get_rabbitmq_exchange_name(),
exchange_type=self.get_rabbitmq_exchange_type(),
durable=True,
)
return channel
def thread_safe_publish(self, message):
try:
random_channel_number = CommonUtils.get_random_number(0, self.get_channel_count() - 1)
channel = self.channels[random_channel_number]
if not channel or channel.is_closed:
channel = self.assign_channel()
if channel:
self.channels[random_channel_number] = channel
self.channels[random_channel_number].basic_publish(
exchange=self.get_rabbitmq_exchange_name(),
routing_key=self.get_rabbitmq_routing_key(),
body=json.dumps(message),
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
)
)
event_key = self.get_event_key()
self.process_data_events(event_key)
except Exception as e:
Log.e(f"Failed to send message to RabbitMQ: {e}")
def process_data_events(self, event_key):
try:
if not self.connection or self.connection.is_closed:
self.connect()
self.connection.process_data_events(time_limit=0)
import time
CacheUtils.set_key_payload(key=event_key, payload=int(TimeUtils.current_milli_time()))
except Exception as e:
Log.e(str(e))
def close_channels(self):
try:
if self.channels:
for key, channel in self.channels.items():
if channel.is_open:
channel.close()
except Exception as e:
Log.e(str(e))
self.channels = {}
@abstractmethod
def get_rabbitmq_routing_key(self):
pass
@abstractmethod
def get_rabbitmq_exchange_name(self):
pass
@abstractmethod
def get_rabbitmq_exchange_type(self):
pass
@abstractmethod
def get_queue_message_type(self):
pass
@abstractmethod
def get_event_key(self):
pass
@abstractmethod
def get_channel_count(self):
pass
@abstractmethod
def get_connection_name(self):
pass
class EmailPublisher(RabbitMQProducer):
__singleton_instance = None
@classmethod
def instance(cls):
# check for the singleton instance
if not cls.__singleton_instance:
cls.__singleton_instance = EmailPublisher()
# return the singleton instance
return cls.__singleton_instance
def get_rabbitmq_routing_key(self):
return 'email.queue'
def get_rabbitmq_exchange_name(self):
return 'email_exchange'
def get_rabbitmq_exchange_type(self):
return "direct"
def get_channel_count(self):
return 5
def get_connection_name(self):
return 'email_connection'
1条答案
按热度按时间f4t66c6m1#
现在我想分享gunicorn工作者之间的rabbitmq连接,我该怎么做?
你不能!**
Gunicorn worker是 * 独立的操作系统进程 *。每个worker都必须有自己到RabbitMQ的连接。
**注意:**Team RabbitMQ监控
rabbitmq-users
邮件列表,仅在StackOverflow上偶尔回答问题。