Есть ли опыт работы с RabbitMQ

Да, у меня есть опыт тестирования систем, использующих RabbitMQ в качестве брокера сообщений. Моя работа была сосредоточена на обеспечении надежности и корректности асинхронной коммуникации между сервисами.

Ключевые направления тестирования:

1. Функциональное тестирование очередей:

   • Доставка сообщений: Проверка, что сообщение, опубликованное в exchange, корректно маршрутизируется и попадает в целевую очередь согласно routing key и типу exchange (direct, fanout, topic).
   • Целостность данных: Валидация того, что тело сообщения (payload) не искажается при передаче.
   • Обработка подтверждений (ACK/NACK): Тестирование сценариев, когда потребитель подтверждает (basic_ack) или отклоняет (basic_nack) сообщение, и проверка его повторной доставки или перемещения в Dead Letter Exchange (DLX).

2. Нагрузочное и стресс-тестирование:

   • Оценка пропускной способности очередей при высокой нагрузке.
   • Проверка поведения системы при заполнении очередей (лимиты x-max-length) и срабатывании политик отбрасывания сообщений.
   • Мониторинг потребления памяти и дискового пространства брокером в ходе длительных тестов.

3. Тестирование отказоустойчивости:

   • Восстановление работы после перезапуска RabbitMQ-ноды.
   • Проверка работы кластеров RabbitMQ (mirrored queues) при отказе одной из нод.
   • Тестирование сетевых разрывов между продюсерами/консьюмерами и брокером.

Пример интеграционного теста на Python с использованием библиотеки pika:

import pika
import json
import pytest

@pytest.fixture
def rabbit_channel():
    """Фикстура для создания подключения и канала к RabbitMQ."""
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_payments', durable=True)
    yield channel
    channel.queue_delete(queue='test_payments')
    connection.close()

def test_payment_message_processing(rabbit_channel):
    """Тест на отправку и получение сообщения о платеже."""
    # 1. Подготовка тестового сообщения
    test_payment = {
        "transaction_id": "txn_12345",
        "amount": 100.50,
        "currency": "USD"
    }

    # 2. Публикация сообщения в очередь
    rabbit_channel.basic_publish(
        exchange='',
        routing_key='test_payments',
        body=json.dumps(test_payment),
        properties=pika.BasicProperties(delivery_mode=2)  # Сообщение persistent
    )

    # 3. Получение сообщения из очереди (имитация работы консьюмера)
    method_frame, header_frame, body = rabbit_channel.basic_get(queue='test_payments')

    # 4. Проверки
    assert method_frame is not None, "Сообщение не было доставлено в очередь"

    received_payment = json.loads(body.decode())
    assert received_payment["transaction_id"] == test_payment["transaction_id"]
    assert received_payment["amount"] == test_payment["amount"]

    # 5. Подтверждение обработки (ACK)
    rabbit_channel.basic_ack(method_frame.delivery_tag)

    # 6. Проверка, что очередь пуста после ACK
    method_frame_after_ack, _, _ = rabbit_channel.basic_get(queue='test_payments')
    assert method_frame_after_ack is None, "Очередь должна быть пуста после подтверждения"

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