VasyOK
Full Member | Редактировать | Профиль | Сообщение | Цитировать | Сообщить модератору
Друзья, всем привет.
Необходима помощь знатоков Python и RabbitMQ.
> Что имеем: три сервера, два из них в яндекс облаке, на одном rabbitmq, на другом скрипт на питоне, третий находится в другом облаке и на нем тоже скрипт на питоне. Скрипты крутятся как сервисы.
> Проблема: Обмен сообщениями происходит со скорость в 1-2 сек, что как мне кажется очень медленно и что это можно ускорить.
RabbitMQ стоит дефолтный на Ubuntu.
Пример функций, отправки/получения сообщений:
Код:
connection = None
channel_pool = None
async def get_connection():
global connection
if connection is None or connection.is_closed:
connection = await aio_pika.connect_robust(f"amqp://admin:pas123456@{mq_ip}/")
return connection
async def get_channel():
global channel_pool
if channel_pool is None or channel_pool.is_closed:
connection = await get_connection()
channel_pool = await connection.channel()
return channel_pool
async def publish(consumer, data=None, action=None, event=None):
global service_uuid, service_name
sender_name = service_name + '.' + service_uuid
arr = {'sender': sender_name, 'consumer': consumer, 'message': {'data': data, 'action': action, 'event': event}}
body = json.dumps(arr)
await set_log(body)
channel = await get_channel()
exchange = await channel.declare_exchange(
"direct", auto_delete=True
)
queue = await channel.declare_queue(
consumer, auto_delete=True
)
await queue.bind(exchange, consumer)
body = bytes(body, "utf-8")
result = await exchange.publish(
aio_pika.Message(body, content_type="text/plain"),
consumer,
)
async def consuming():
global service_name, demons_server_ip
connection = await get_connection()
async with connection:
queue_name = service_name + '.' + service_uuid
channel: aio_pika.abc.AbstractChannel = await connection.channel()
queue: aio_pika.abc.AbstractQueue = await channel.declare_queue(queue_name, auto_delete=True)
async with queue.iterator() as queue_iter:
async for message in queue_iter:
data = message.body
asyncio.create_task(callback(data))
await message.ack()
|
Возможно ли обмен сообщениями ускорить? |
