Тестирование очередей сообщений: Асинхронные системы и Event-Driven архитектура

Очереди сообщений являются фундаментальными для современных распределенных систем, обеспечивая асинхронную коммуникацию и развязку сервисов. Это всеобъемлющее руководство охватывает стратегии тестирования для очередей сообщений, фокусируясь на SQS, Azure (как обсуждается в Serverless Testing Guide: AWS Lambda and Azure Functions) Queue и общих паттернах, таких как порядок сообщений, идемпотентность и обработка poison messages.

Понимание вызовов тестирования очередей сообщений

Тестирование систем очередей сообщений требует решения нескольких уникальных вызовов:

Асинхронное поведение: Сообщения обрабатываются в конечном итоге, а не немедленно
Порядок сообщений: Гарантии FIFO vs доставка хотя-бы-один-раз
Идемпотентность: Грамотная обработка дублирующихся сообщений
Время видимости: Управление блокировками сообщений и повторной доставкой
Poison messages: Обнаружение и обработка сообщений, которые постоянно терпят неудачу
Пакетная обработка: Тестирование сценариев высокой пропускной способности

Настройка тестирования SQS

Локальный SQS с LocalStack

# docker-compose.yml
version: '3.8'
services:
  localstack:
    image: localstack/localstack
    ports:
      - "4566:4566"
    environment:
      - SERVICES=sqs
      - DEBUG=1
      - DEFAULT_REGION=us-east-1
    volumes:
      - "./localstack:/var/lib/localstack"

Создание и тестирование очередей:

// sqs-setup.test.js
const AWS (как обсуждается в [Cross-Platform Mobile Testing: Strategies for Multi-Device Success](/blog/cross-platform-mobile-testing)) = require('aws-sdk');

const sqs = new AWS.SQS({
  endpoint: 'http://localhost:4566',
  region: 'us-east-1',
  accessKeyId: 'test',
  secretAccessKey: 'test'
});

describe('Настройка очереди SQS', () => {
  let queueUrl;

  beforeAll(async () => {
    const result = await sqs.createQueue({
      QueueName: 'test-queue',
      Attributes: {
        DelaySeconds: '0',
        MessageRetentionPeriod: '86400', // 1 день
        VisibilityTimeout: '30'
      }
    }).promise();

    queueUrl = result.QueueUrl;
  });

  test('должна создать очередь успешно', async () => {
    const result = await sqs.getQueueAttributes({
      QueueUrl: queueUrl,
      AttributeNames: ['All']
    }).promise();

    expect(result.Attributes.QueueArn).toBeDefined();
    expect(result.Attributes.VisibilityTimeout).toBe('30');
  });

  test('должна отправить сообщение в очередь', async () => {
    const message = {
      orderId: '123',
      userId: 'user-456',
      items: [{ id: 'item1', quantity: 2 }]
    };

    const result = await sqs.sendMessage({
      QueueUrl: queueUrl,
      MessageBody: JSON.stringify(message)
    }).promise();

    expect(result.MessageId).toBeDefined();
  });

  test('должна получить сообщение из очереди', async () => {
    const result = await sqs.receiveMessage({
      QueueUrl: queueUrl,
      MaxNumberOfMessages: 1,
      WaitTimeSeconds: 5
    }).promise();

    expect(result.Messages).toBeDefined();
    expect(result.Messages.length).toBeGreaterThan(0);

    const message = JSON.parse(result.Messages[0].Body);
    expect(message.orderId).toBe('123');
  });

  afterAll(async () => {
    await sqs.deleteQueue({ QueueUrl: queueUrl }).promise();
  });
});

Тестирование времени видимости сообщений

Время видимости определяет как долго сообщения невидимы для других потребителей после получения:

// visibility-timeout.test.js
describe('Время видимости', () => {
  let queueUrl;

  beforeAll(async () => {
    const result = await sqs.createQueue({
      QueueName: 'visibility-test-queue',
      Attributes: {
        VisibilityTimeout: '5' // 5 секунд
      }
    }).promise();

    queueUrl = result.QueueUrl;
  });

  test('сообщение должно быть невидимым во время таймаута', async () => {
    // Отправить сообщение
    await sqs.sendMessage({
      QueueUrl: queueUrl,
      MessageBody: JSON.stringify({ id: 1 })
    }).promise();

    // Первый потребитель получает сообщение
    const result1 = await sqs.receiveMessage({
      QueueUrl: queueUrl,
      MaxNumberOfMessages: 1
    }).promise();

    expect(result1.Messages.length).toBe(1);

    // Второй потребитель немедленно пытается получить
    const result2 = await sqs.receiveMessage({
      QueueUrl: queueUrl,
      MaxNumberOfMessages: 1
    }).promise();

    // Не должен получить то же сообщение (оно невидимо)
    expect(result2.Messages || []).toHaveLength(0);
  });

  test('сообщение должно появиться снова после истечения таймаута', async () => {
    await sqs.sendMessage({
      QueueUrl: queueUrl,
      MessageBody: JSON.stringify({ id: 2 })
    }).promise();

    // Получить но не удалять
    const result1 = await sqs.receiveMessage({
      QueueUrl: queueUrl
    }).promise();

    expect(result1.Messages.length).toBe(1);

    // Подождать истечения времени видимости
    await new Promise(resolve => setTimeout(resolve, 6000));

    // Сообщение должно быть снова доступно
    const result2 = await sqs.receiveMessage({
      QueueUrl: queueUrl
    }).promise();

    expect(result2.Messages.length).toBe(1);
    expect(JSON.parse(result2.Messages[0].Body).id).toBe(2);
  });

  afterAll(async () => {
    await sqs.deleteQueue({ QueueUrl: queueUrl }).promise();
  });
});

Механизмы повторов и очереди мертвых сообщений

Тестирование логики повторов и конфигурации DLQ:

// dlq-test.js
describe('Очередь мертвых сообщений', () => {
  let mainQueueUrl, dlqUrl;

  beforeAll(async () => {
    // Создать DLQ
    const dlqResult = await sqs.createQueue({
      QueueName: 'test-dlq'
    }).promise();

    dlqUrl = dlqResult.QueueUrl;

    // Получить DLQ ARN
    const dlqAttrs = await sqs.getQueueAttributes({
      QueueUrl: dlqUrl,
      AttributeNames: ['QueueArn']
    }).promise();

    const dlqArn = dlqAttrs.Attributes.QueueArn;

    // Создать главную очередь с конфигурацией DLQ
    const mainResult = await sqs.createQueue({
      QueueName: 'test-main-queue',
      Attributes: {
        VisibilityTimeout: '2',
        RedrivePolicy: JSON.stringify({
          deadLetterTargetArn: dlqArn,
          maxReceiveCount: '3'
        })
      }
    }).promise();

    mainQueueUrl = mainResult.QueueUrl;
  });

  test('сообщение должно переместиться в DLQ после максимума повторов', async () => {
    // Отправить сообщение в главную очередь
    await sqs.sendMessage({
      QueueUrl: mainQueueUrl,
      MessageBody: JSON.stringify({ id: 'poison-message' })
    }).promise();

    // Получить и не удалять 3 раза (симулируя сбои)
    for (let i = 0; i < 3; i++) {
      await sqs.receiveMessage({
        QueueUrl: mainQueueUrl,
        WaitTimeSeconds: 1
      }).promise();

      // Подождать истечения времени видимости
      await new Promise(resolve => setTimeout(resolve, 2500));
    }

    // Сообщение должно теперь быть в DLQ
    const dlqResult = await sqs.receiveMessage({
      QueueUrl: dlqUrl,
      WaitTimeSeconds: 5
    }).promise();

    expect(dlqResult.Messages).toBeDefined();
    expect(dlqResult.Messages.length).toBe(1);

    const message = JSON.parse(dlqResult.Messages[0].Body);
    expect(message.id).toBe('poison-message');
  });

  afterAll(async () => {
    await sqs.deleteQueue({ QueueUrl: mainQueueUrl }).promise();
    await sqs.deleteQueue({ QueueUrl: dlqUrl }).promise();
  });
});

Тестирование очереди FIFO

Очереди FIFO гарантируют порядок сообщений и обработку ровно-один-раз:

// fifo-queue.test.js
describe('Очередь FIFO', () => {
  let queueUrl;

  beforeAll(async () => {
    const result = await sqs.createQueue({
      QueueName: 'test-queue.fifo',
      Attributes: {
        FifoQueue: 'true',
        ContentBasedDeduplication: 'true'
      }
    }).promise();

    queueUrl = result.QueueUrl;
  });

  test('должна поддерживать порядок сообщений', async () => {
    const messages = ['первое', 'второе', 'третье', 'четвертое', 'пятое'];

    // Отправить сообщения по порядку
    for (const msg of messages) {
      await sqs.sendMessage({
        QueueUrl: queueUrl,
        MessageBody: msg,
        MessageGroupId: 'test-group'
      }).promise();
    }

    // Получить сообщения
    const receivedMessages = [];

    for (let i = 0; i < messages.length; i++) {
      const result = await sqs.receiveMessage({
        QueueUrl: queueUrl,
        MaxNumberOfMessages: 1
      }).promise();

      if (result.Messages && result.Messages.length > 0) {
        receivedMessages.push(result.Messages[0].Body);

        // Удалить сообщение
        await sqs.deleteMessage({
          QueueUrl: queueUrl,
          ReceiptHandle: result.Messages[0].ReceiptHandle
        }).promise();
      }
    }

    // Проверить что порядок поддерживается
    expect(receivedMessages).toEqual(messages);
  });

  test('должна дедуплицировать сообщения', async () => {
    const message = 'тестовое-сообщение-дубликат';

    // Отправить то же сообщение дважды в окне дедупликации
    await sqs.sendMessage({
      QueueUrl: queueUrl,
      MessageBody: message,
      MessageGroupId: 'dedup-group'
    }).promise();

    await sqs.sendMessage({
      QueueUrl: queueUrl,
      MessageBody: message,
      MessageGroupId: 'dedup-group'
    }).promise();

    // Получить сообщения
    const result1 = await sqs.receiveMessage({
      QueueUrl: queueUrl
    }).promise();

    expect(result1.Messages.length).toBe(1);

    // Удалить первое сообщение
    await sqs.deleteMessage({
      QueueUrl: queueUrl,
      ReceiptHandle: result1.Messages[0].ReceiptHandle
    }).promise();

    // Попытаться получить снова - должно быть пусто (дубликат был проигнорирован)
    const result2 = await sqs.receiveMessage({
      QueueUrl: queueUrl,
      WaitTimeSeconds: 2
    }).promise();

    expect(result2.Messages || []).toHaveLength(0);
  });

  afterAll(async () => {
    await sqs.deleteQueue({ QueueUrl: queueUrl }).promise();
  });
});

Тестирование идемпотентного потребителя

Обеспечение правильной обработки дублирующихся сообщений потребителями:

// idempotent-consumer.test.js
class IdempotentOrderProcessor {
  constructor() {
    this.processedOrders = new Set();
    this.orders = [];
  }

  async processOrder(message) {
    const order = JSON.parse(message.Body);

    // Проверить был ли уже обработан
    if (this.processedOrders.has(order.id)) {
      console.log(`Заказ ${order.id} уже обработан, пропускаем`);
      return { processed: false, duplicate: true };
    }

    // Обработать заказ
    this.orders.push(order);
    this.processedOrders.add(order.id);

    return { processed: true, duplicate: false };
  }
}

describe('Идемпотентный потребитель', () => {
  let processor;
  let queueUrl;

  beforeAll(async () => {
    const result = await sqs.createQueue({
      QueueName: 'idempotent-test-queue'
    }).promise();

    queueUrl = result.QueueUrl;
    processor = new IdempotentOrderProcessor();
  });

  test('должен обработать сообщение только один раз', async () => {
    const order = { id: 'order-123', amount: 100 };

    // Отправить сообщение дважды (симулируя дубликат)
    await sqs.sendMessage({
      QueueUrl: queueUrl,
      MessageBody: JSON.stringify(order)
    }).promise();

    await sqs.sendMessage({
      QueueUrl: queueUrl,
      MessageBody: JSON.stringify(order)
    }).promise();

    // Получить и обработать оба сообщения
    const result1 = await sqs.receiveMessage({
      QueueUrl: queueUrl
    }).promise();

    const processed1 = await processor.processOrder(result1.Messages[0]);
    expect(processed1.processed).toBe(true);
    expect(processed1.duplicate).toBe(false);

    const result2 = await sqs.receiveMessage({
      QueueUrl: queueUrl,
      WaitTimeSeconds: 2
    }).promise();

    const processed2 = await processor.processOrder(result2.Messages[0]);
    expect(processed2.processed).toBe(false);
    expect(processed2.duplicate).toBe(true);

    // Проверить что был добавлен только один заказ
    expect(processor.orders.length).toBe(1);
  });

  afterAll(async () => {
    await sqs.deleteQueue({ QueueUrl: queueUrl }).promise();
  });
});

Лучшие практики тестирования очередей сообщений

Чеклист тестирования

Тестировать отправку и получение сообщений
Проверить поведение времени видимости
Тестировать механизмы повторов и конфигурацию DLQ
Валидировать гарантии порядка FIFO
Тестировать дедупликацию сообщений
Реализовать идемпотентных потребителей
Грамотно обрабатывать poison messages
Тестировать сценарии пакетной обработки
Проверить атрибуты и метаданные сообщений
Тестировать long-polling vs short-polling
Мониторить метрики очереди (глубина, возраст и т.д.)
Тестировать обработку ошибок и логирование

Сравнение типов очередей

Характеристика	Стандартная очередь	FIFO очередь
Порядок	Лучшие усилия	Строгий FIFO
Доставка	Хотя-бы-один-раз	Ровно-один-раз
Пропускная способность	Неограниченная	300 TPS (пакеты: 3000)
Дедупликация	Нет	Да (окно 5 минут)
Случай использования	Высокая пропускная способность, порядок не критичен	Порядок важен, без дубликатов

Заключение

Эффективное тестирование очередей сообщений требует всеобъемлющего покрытия асинхронного поведения, порядка сообщений, идемпотентности, логики повторов и обработки poison messages. Реализуя тщательные тесты для времени видимости, конфигурации DLQ, гарантий FIFO и пакетной обработки, вы можете обеспечить надежные архитектуры, управляемые сообщениями.

Ключевые выводы:

Всегда реализовывать идемпотентных потребителей для доставки хотя-бы-один-раз
Тестировать поведение времени видимости для предотвращения потери сообщений
Настраивать и тестировать DLQ для обработки poison messages
Использовать FIFO очереди когда порядок критичен
Реализовывать правильную логику повторов с экспоненциальной задержкой
Мониторить метрики очереди в продакшене

Надежное тестирование очередей сообщений создает уверенность в асинхронных системах и предотвращает потерю данных и сбои обработки.