Обмен сообщениями «точка-точка» с помощью Amazon SQS

Введение

Шаблон обмена сообщениями «точка-точка» — это широко используемая модель коммуникации в современных веб- и облачных архитектурах. Он предназначен для обеспечения асинхронного взаимодействия между различными компонентами, например, бессерверными функциями или микросервисами, позволяя им обмениваться сообщениями, не требуя немедленного ответа.

В этом шаблоне компонент, отправляющий сообщение, называется производителем, а компонент, принимающий и обрабатывающий сообщение, — потребителем. Производитель и потребитель могут располагаться как в одной системе, так и в разных, что делает этот подход к коммуникации гибким и масштабируемым.

Подобно тому, как электронные письма доставляются получателям, сообщения отправляются от отправителя конкретному получателю. Это обеспечивает эффективную и надёжную коммуникацию даже в сложных распределённых системах. Обычно это используется в сценариях, когда отправитель точно знает, какой получатель должен получить сообщение, но ему не требуется немедленный ответ.

Шаблон обмена сообщениями «точка-точка» эффективно облегчает взаимодействие и координацию между компонентами, повышая общую производительность, надежность и масштабируемость современных веб- и облачных архитектур.

Что мы построим

В этом пошаговом руководстве мы реализуем этот шаблон с использованием двух функций AWS Lambda и очереди Amazon SQS.

В этом пошаговом руководстве мы реализуем простой пример с использованием двух функций AWS Lambda и очереди Amazon SQS.

Вы создадите пример с использованием TypeScript и AWS Cloud Development Kit (AWS CDK).

Пример будет состоять из трех компонентов:

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

Помимо реализации этого шаблона, мы также расскажем о возможностях AWS Cloud Development Kit (CDK), позволяющих определить всю инфраструктуру как код. Если вы хотите узнать больше об AWS CDK, ознакомьтесь с руководством разработчика AWS CDK.

К концу этого руководства вы будете иметь полное представление об отдельных компонентах двухточечного обмена сообщениями на основе очередей, успешно реализовали асинхронную связь между двумя лямбда-функциями с помощью SQS и приобретёте практический опыт построения инфраструктуры. Программирование с использованием CDK

Но прежде чем приступить к кодированию, давайте кратко рассмотрим преимущества и недостатки асинхронной модели обмена сообщениями «точка-точка».

Преимущества и недостатки парадигмы асинхронного обмена сообщениями «точка-точка»

Преимущества

Слабое связывание: асинхронный шаблон обмена сообщениями «точка-точка» способствует слабому связыванию между приложениями, позволяя им взаимодействовать независимо, без необходимости тесной интеграции. Эта гибкость упрощает масштабирование и модификацию отдельных компонентов без ущерба для всей системы.
Масштабируемость: Этот шаблон обеспечивает эффективное горизонтальное масштабирование, поскольку позволяет добавлять несколько приложений-потребителей для асинхронной обработки рабочих нагрузок. Это позволяет системе эффективнее обрабатывать большие объёмы одновременных сообщений и запросов.
Надежность: при асинхронном обмене сообщениями, если сообщение не доставлено или не обработано, его можно повторить или отправить в очередь ошибок для последующей обработки, что повышает надежность системы.
Отказоустойчивость: асинхронный обмен сообщениями обеспечивает отказоустойчивость, разделяя отправителей и получателей сообщений. В случае сбоя одного приложения или компонента сообщения могут быть сохранены для последующей обработки после восстановления работы системы, что гарантирует отсутствие потери данных.

Недостатки

Сложность: Реализация асинхронного шаблона обмена сообщениями «точка-точка» может оказаться более сложной по сравнению с другими шаблонами интеграции и требует дополнительной логики обработки сообщений.
Зависимости сообщений и возможное дублирование: Управление зависимостями между сообщениями и обеспечение корректного удаления сообщений может быть сложной задачей в асинхронной системе обмена сообщениями. Для решения таких потенциальных проблем, как упорядочивание сообщений, дублирование сообщений и зависимости при их обработке, требуется тщательное проектирование и реализация.
Увеличенная задержка: асинхронный обмен сообщениями приводит к задержке между отправкой сообщения и получением ответа, поскольку обработка сообщений может занимать больше времени. Эта задержка может повлиять на взаимодействие в реальном времени и может быть неподходящей для приложений, требующих немедленной обратной связи.

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

Но теперь давайте начнем руководство и узнаем, как реализовать этот шаблон с помощью AWS Lambda и Amazon SQS.

Примечание о затратах ресурсов при кодировании: в этом руководстве используется только минимальный объем ресурсов, все из которых включены в бесплатный уровень, предлагаемый AWS в течение первых 12 месяцев после создания каждой учетной записи:

Несколько килобайт кода будут храниться в Amazon S3, который предоставляет 5 ГБ бесплатного хранилища.
Мы несколько раз звоним в SQS, которая предлагает 1 миллион бесплатных запросов в месяц.
Мы вызовем две функции в AWS Lambda, которая также предлагает 1 миллион бесплатных вызовов в месяц.

Так что, следуя пошаговому руководству, вы точно останетесь на бесплатном уровне. Я также добавил в конце раздел, который поможет вам удалить все ресурсы, созданные в ходе этого руководства.

Предпосылки

Если у вас установлены CLI и CDK, вы увидите версии своих установок. В противном случае появится сообщение об ошибке, сообщающее о том, что команда не найдена.

Прежде чем приступить к выполнению этого урока, вам понадобится следующее:

аккаунт AWS
Комплект разработки AWS Cloud (AWS CDK)

Для этого руководства требуются как минимум AWS CLI и AWS CDK версии 2.

Определить версию AWS CLI можно, выполнив следующую команду в терминале:

aws --version

И версию AWS CDK, выполнив следующую команду:

cdk --version

Шаг 1 — Создание приложения CDK

1.1 – Запуск программы CDK

Сначала вам необходимо позаботиться о первоначальной настройке и позволить CDK автоматически сгенерировать код, который будет использоваться для сборки проекта.

В командной строке создайте каталог для проекта, перейдите в него и инициализируйте проект:

mkdir point-to-point-example
cd point-to-point-example
cdk init app --language typescript

Это создает основную файловую структуру проекта и устанавливает необходимые пакеты.

Сейчас самое время открыть проект в вашей любимой среде IDE и взглянуть на сгенерированный проект, а именно на следующие два файла:

point-to-point-example/
├── bin/
│ └── point-to-point-example.ts
├── lib/
│ └── point-to-point-example-stack.ts
└── ...

Файл bin\point-to-point-example.ts Точка входа для программы CDK. Нам не придётся снова рассматривать этот файл в этом руководстве.

Файл lib\point-to-point-example-stack.ts Именно здесь определяется основной стек программы. Именно в этом файле мы проводим большую часть времени.

По мере продвижения по этому уроку мы познакомим класс с инфраструктурой нашего приложения. PointToPointExampleStack В этот файл мы добавим. Откройте его в редакторе и посмотрите:

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
// import * as sqs from 'aws-cdk-lib/aws-sqs';
export class PointToPointExampleStack extends cdk.Stack {
constructor(scope: Construct, id: string, props?: cdk.StackProps) {
super(scope, id, props);
// The code that defines your stack goes here
// ...
}
}

1.2 – Начальная настройка среды

При первом развертывании приложения AWS CDK в среде, то есть в конкретной учётной записи AWS и регионе, необходимо загрузить среду со стеком CDKToolkit. Этот стек развёртывает контейнер S3, который используется для хранения шаблонов и ресурсов в процессе развертывания приложения.

Чтобы загрузить среду, убедитесь, что вы все еще находитесь в каталоге примера «точка-точка», и выполните следующую команду:

cdk bootstrap

Это займет некоторое время и будет завершено, когда вы увидите следующий вывод в командной строке:

✅ Environment aws://[account_id]/[region] bootstrapped.

1.3 – Развертывание начального стека

Чтобы развернуть стек, убедитесь, что вы все еще находитесь в каталоге примера «точка-точка», и выполните следующую команду:

cdk deploy

Эта команда компилирует приложение в шаблон AWS CloudFormation и развертывает его в вашей учетной записи AWS.

Теперь откройте сведения о стеке и изучите его. На вкладке «Ресурсы» вы увидите, что ничего не развернуто, кроме ресурса CDKMetadata, который содержит лишь некоторые начальные данные конфигурации нашего стека.

Шаг 2 — Создание очереди SQS

Как только все настроено, вы можете создать свой первый реальный ресурс — очередь SQS.

2.1 – Добавить очередь в стек

Для начала, файл lib\point-to-point-example-stack.ts Открыть.

В зависимости от версии шаблона CDK, в него уже могли быть добавлены некоторые операторы импорта, включая aws-cdk-lib/aws-sqs в комментариях. Если это так, удалите символы комментария // перед import * as sqs from 'aws-cdk-lib/aws-sqs', в противном случае добавьте их самостоятельно, чтобы первые три строки файла выглядели следующим образом:

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as sqs from 'aws-cdk-lib/aws-sqs';

Теперь комментарии в классе. PointToPointExampleStack Удалите и замените следующий код и сохраните изменения:

const queue = new sqs.Queue(this, 'Queue', {
queueName: 'MyQueue',
});

2.2 – Развертывание очереди

Чтобы развернуть очередь, выполните следующую команду в каталоге вашего проекта:

cdk deploy

Теперь вы можете отслеживать процесс развертывания в выходных данных терминала, он должен быть завершен через минуту или две.

После завершения развертывания перейдите в список очередей SQS в консоли управления AWS и найдите очередь под названием SqsTutorialStack-Queue:

2.3 – Проверка очереди «Развернуто»

Перейдите на панель управления SQS в консоли управления AWS и просмотрите список очередей. Если вы не можете найти свою очередь, убедитесь, что вы находитесь в правильном регионе.

Шаг 3 — Создайте функцию Producer и подключите ее к очереди.

3.1 – Напишите исходный код функции-генератора

Мы будем использовать JavaScript для лямбда-функции и сохраним исходный код в отдельной папке внутри приложения CDK. Позже мы будем ссылаться на эту папку из нашего стека CDK, чтобы он мог автоматически упаковать, загрузить и развернуть содержимое в AWS.

Убедитесь, что вы находитесь в каталоге проекта, и создайте следующие каталоги и сам файл:

mkdir lambda-src
mkdir lambda-src/producer
touch lambda-src/producer/send_message_to_sqs.js

Структура каталогов вашего проекта теперь должна выглядеть следующим образом:

point-to-point-example/
├── bin/
│ └── point-to-point-example.ts
├── lambda-src/
│ └── producer/
│ └── send_message_to_sqs.js
├── lib/
│ └── point-to-point-example-stack.ts
└── ...

Создан новый файл. lambda-src/producer/send_message_to_sqs.js Откройте его в редакторе и добавьте следующий код:

const sqs = require("@aws-sdk/client-sqs");
// Create an Amazon SQS service client
const client = new sqs.SQSClient();
exports.handler = async (event, context) => {
// Create a message, including the function's name from
// its execution context
const myMessage = 'Hello World from ' + context.functionName
// Get the queue URL from the execution environment
const queueUrl = process.env.SQSQueueUrl
// Create a SendMessageCommand containing the message
// and the queue URL
const command = new sqs.SendMessageCommand({
MessageBody: JSON.stringify({ message: myMessage }),
QueueUrl: queueUrl
});
// Send the message and return the result 
const result = await client.send(command);
return result;
}

Это простой обработчик лямбда-функции, который отправляет сообщение в нашу очередь SQS и возвращает результат.

Обратите внимание: поскольку мы не знаем, какую очередь использовать при построении этой функции, или можем захотеть изменить её позже, не изменяя код функции, мы получаем URL-адрес очереди из среды выполнения. Таким образом, мы можем определить его динамически во время развёртывания лямбда-функции в стеке CDK.

3.2 – Добавить функцию Producer в стек

Чтобы создать настоящую лямбда-функцию в AWS, необходимо добавить её в стек. Для этого откройте lib/point-to-point-example-stack.ts.

Сначала добавьте еще один оператор импорта в начало файла:

import * as lambda from 'aws-cdk-lib/aws-lambda';

Затем добавьте следующий фрагмент внутрь конструктора, под очередью:

const producerFunction = new lambda.Function(this, 'Producer', {
functionName: 'Producer',
runtime: lambda.Runtime.NODEJS_18_X,
code: lambda.Code.fromAsset('lambda-src/producer'),
handler: 'send_message_to_sqs.handler',
environment: {
SQSQueueUrl: queue.queueUrl,
}
});

Это создает функцию, пакеты и содержимое каталога lambda-src/producer, а также настраивает функцию на использование обработчика, определенного в send_message_to_sqs.js, при вызове.

Он также динамически устанавливает переменную среды SQSQueueUrl на фактический URL-адрес очереди SQS, чтобы к ней можно было получить доступ из функции с помощью process.env.SQSQueueUrl.

3.3 – Лицензирование исполнения продюсера

Каждая функция Lambda требует нескольких базовых разрешений IAM, которые CDK предоставляет по умолчанию с помощью базовой роли выполнения.

Однако нашей функции-производителю требуются дополнительные разрешения для отправки сообщений в очередь. К счастью, в CDK нам не нужно вручную управлять политиками IAM. Мы можем использовать метод grantSendMessages очереди SQS с нашей лямбда-функцией.

Для этого добавьте следующую строку под producerFunction:

queue.grantSendMessages(producerFunction);

3.4 – Развертывание функции производителя

Чтобы развернуть функцию вместе с ее исходным кодом, выполните следующую команду в каталоге проекта:

cdk deploy

Это подготавливает развертывание любых новых или измененных ресурсов.

Если какие-либо изменения в вашем стеке потребуют создания или изменения разрешений IAM, CDK предложит вам повторно просмотреть эти изменения из соображений безопасности.

На изображении ниже вы можете видеть, что CDK хочет внести три изменения, которые повлияют на разрешения IAM:

Разрешить службе Lambda (lambda.amazon.aws) взять на себя роль исполнения, прикрепленную к функции Producer.
Добавьте необходимые разрешения напрямую, чтобы разрешить исполнительной роли получать доступ и отправлять сообщения в очередь SQS.
Добавьте управляемую политику с начальными разрешениями для роли исполнения.

Введите y для подтверждения, и CDK развернет новую функцию Lambda вместе с ее ролью IAM.

3.5 – Проверка производительности производителя

После запуска функции перейдите к списку функций на панели управления AWS Lambda.

Имя функции совпадает с тем, что мы определили в стеке: Producer. Если вы не можете найти функцию, убедитесь, что вы находитесь в правильном регионе.

Щелкните имя функции, чтобы открыть ее подробную информацию и перейти к разделу исходного кода.

Папка проводника слева содержит файл send_message_to_sqs.js.

Дважды щелкните по имени файла, чтобы открыть его исходный код. Вы увидите, что это точная копия файла, который мы создали в папке lambda-src/producer в нашем проекте CDK.

Теперь давайте настроим тестовое событие.

Для этого нажмите кнопку “Тест”. Если вы ещё не настроили тестовое событие, откроется страница «Конфигурация теста». Введите название события, например «test-event», и нажмите «Сохранить».

После настройки тестового события снова нажмите кнопку Тест, что приведет к непосредственному вызову лямбда-функции.

Это создаст вкладку «Результаты выполнения», содержащую ответ:

{
"$metadata": {
"httpStatusCode": 200,
"requestId": "cb033b2a-1234-5678-9012-66188359d280",
"attempts": 1,
"totalRetryDelay": 0
},
"MD5OfMessageBody": "b5357af0c1c816b2d41275537cc0d1af",
"MessageId": "4a76e538-1234-5678-9012-749f0f4b9294"
}

Помните фрагмент кода в лямбда-функции?

const result = await client.send(command);
return result;

Объект JSON в приведенных выше результатах выполнения — это именно то, что SQS вернул в ответ на client.send(command), включая интересную информацию, например “httpStatusCode”: 200, что означает, что очередь успешно получила сообщение.

3.6 – Извлечение сообщений из SQS вручную

В настоящее время очередь не подключена к потребителю, поэтому сообщение не будет получено. Однако мы можем получить сообщение вручную.

Перейдите на панель управления SQS, щелкните очередь, чтобы открыть страницу с ее сведениями, и нажмите кнопку «Отправить и получить сообщения».

На следующей странице перейдите в раздел «Получение сообщений» и нажмите кнопку «Опрос сообщений», которая должна отобразить все сообщения, которые в данный момент передаются, включая сообщения, отправленные вашей функцией-конструктором при ее тестировании:

Нажмите на идентификатор сообщения, чтобы открыть его. Текст сообщения должен содержать следующее:

{"message":"Hello World from Producer"}

Поздравляем! Вы успешно развернули и протестировали лямбда-функцию, которая отправляет сообщения в очередь SQS.

Затем мы создаем функцию-потребитель и настраиваем ее на автоматический вызов при поступлении нового сообщения в очередь.

Шаг 4 — Создайте функцию Consumer и подключите ее к очереди.

4.1 – Напишите исходный код функции-потребителя

Код функции потребителя хранится в собственной папке внутри lambda-src, чтобы CDK мог автоматически упаковать его и загрузить в новую функцию Lambda.

Убедитесь, что вы находитесь в каталоге проекта (не в bin/ или lib/) и создайте следующий каталог с исходным файлом:

mkdir lambda-src/consumer
touch lambda-src/consumer/receive_message_from_sqs.js

Откройте lambda-src/consumer/receive_message_from_sqs.js в редакторе и введите следующий код:

exports.handler = async (event) => {
// Retrieve the message from the event and log it to the console
const message = event.Records[0].body;
console.log(message);
}

Это очень простой обработчик лямбда-функции, который открывает сообщение SQS и выводит его в отчет.

4.2 – Добавить функцию Consumer в стек

Добавьте следующий фрагмент внутрь производителя, под queue.grantSendMessages(producerFunction):

 const consumerFunction = new lambda.Function(this, 'Consumer', {
functionName: 'Consumer',
runtime: lambda.Runtime.NODEJS_18_X,
code: lambda.Code.fromAsset('lambda-src/consumer'),
handler: 'receive_message_from_sqs.handler'
});

Эта функция потребителя создает пакеты и содержимое каталога lambda-src/consumer и настраивает функцию на использование обработчика, определенного в receive_message_from_sqs.js, при вызове.

4.3 – Добавьте очередь SQS как источник событий в The Consumer

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

Сначала добавьте еще один оператор импорта в начало файла:

import { SqsEventSource } from 'aws-cdk-lib/aws-lambda-event-sources';

Затем добавьте следующие две строки под определением customerFunction:

consumerFunction.addEventSource(new SqsEventSource(queue));
queue.grantConsumeMessages(consumerFunction);

4.4 – Развертывание функции потребителя

Чтобы развернуть функцию вместе с ее исходным кодом, выполните следующую команду в каталоге проекта:

cdk deploy

Это подготавливает развертывание любых новых или измененных ресурсов и снова предлагает вам повторно проверить новые разрешения IAM.

Введите y для подтверждения, и CDK выполнит новую лямбда-функцию вместе с ее ролью выполнения и присоединит ее к очереди SQS.

После завершения развёртывания всё будет готово и подключено. Посмотрим, как всё заработает.

Шаг 5 – Сквозное тестирование приложения

Для тестирования приложения мы вручную вызываем функцию-производителя для отправки ещё одного сообщения в очередь. Затем мы проверяем журналы выполнения функции-потребителя, чтобы убедиться, что она автоматически запустила и обработала сообщение.

5.1 – Вызов функции Producer

Выполните шаги, описанные в разделе 3.5 – Тестирование функции-производителя, чтобы снова вызвать функцию-производителя. Вот краткое описание:

Перейдите к списку функций на панели управления AWS Lambda.
Откройте функцию Producer, щелкнув по ее названию.
Перейти к разделу исходного кода.
Нажмите кнопку Тест.
Убедитесь, что функция выполняется успешно.

5.2 – Проверьте отчет о потребительских показателях

Вернитесь к списку функций на панели управления AWS Lambda и на этот раз откройте функцию Consumer.

Прокрутите вниз, откройте вкладку под названием «Монитор», внутри этой вкладки откройте вкладку «Журналы» и просмотрите раздел «Недавние вызовы», где вы найдете ссылку на последний LogStream функции.

После этого нажмите на ссылку «Log Flow». Отобразится сама запись журнала, что может показаться немного странным, если вы не привыкли к такому формату.

Просто найдите запись прямо посередине между этими двумя, начинающуюся с START RequestId: … и END Request ID: ….

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

Результат

Поздравляем! Вы создали всё необходимое для настройки асинхронного обмена сообщениями «точка-точка» между двумя лямбда-функциями с помощью SQS.

SQS позволяет делать гораздо больше, например, классифицировать сообщения, контролировать их порядок и автоматически отправлять некорректные сообщения в очередь с истекающим сроком доставки (DLQ). Подробнее см. в руководстве разработчика Amazon SQS.

Обмен сообщениями «точка-точка» с Amazon SQS