使用 Amazon SQS 进行点对点消息传递
教程结束后及时清理则免费（清理步骤见下文）
注册 / 登录 亚马逊云科技账户
Amazon 云开发工具包
本教程的代码可在 GitHub 上获取
简介
点对点消息传递模式是现代 Web 和云架构中常用的通信模型。这种模式旨在实现不同组件（例如无服务器函数或微服务）之间的异步交互，使它们能够彼此之间交换消息而无需立即响应。
在这种模式中，发送消息的组件叫做生产者，而接收和处理消息的组件叫做消费者。生产者和消费者可以位于相同系统或不同系统中，所以这种通信方法既灵活又可扩展。
类似于将电子邮件传递给各个收件人，消息也是从生产者发送至特定的消费者。即使是在复杂的分布式系统中，这种方法也可以实现高效可靠的通信。此方法的常见使用场景是生产者明确知道哪个消费者需要接收消息，但生产者不需要立即收到响应。
点对点消息传递模式有效地促进了组件之间的通信和协作，提高了现代 Web 和云架构的整体性能、可靠性和可扩展性。
构建内容
- 在这次逐步教程中，我们将使用两个 Amazon Lambda 函数和一个 Amazon SQS 队列来实现此模式
- 在这次逐步教程中，我们将使用两个 Amazon Lambda 函数和一个 Amazon SQS 队列来实现一个简单的示例。
- 您将使用 TypeScript 和 Amazon 云开发工具包 (Amazon CDK) 来构建这个示例。
- 这个示例将由三个部分组成：
- 可以向消费者发送消息的生产者
- 可以从生产者接收消息的消费者
- 在生产者和消费者之间建立通信通道的消息队列
除了实现此模式之外，我们还将重点介绍 Amazon 云开发工具包 (CDK) 将整个基础设施定义为代码的强大功能。如果您想了解有关 Amazon CDK 的更多信息，请查看 Amazon CDK 开发人员指南。
学完本教程，您将会深刻了解基于队列的点对点消息传递的各个部分、使用 SQS 成功实现两个 Lambda 函数之间的异步通信，并获得一些使用 CDK 将基础设施作为代码构建的实践经验。
不过，在开始编写代码之前，让我们快速了解一下异步点对点消息传递模式的优缺点。
异步点对点消息传递模式的优缺点
优点
- 弱耦合：异步点对点消息传递模式促进了应用程序之间的弱耦合，使各个应用程序能够独立通信，无需紧密集成。这种灵活性使得在不影响整个系统的情况下更容易地扩展和修改单个组件。
- 可扩展性：这种模式有利于高效的横向扩展，因为可以添加多个消费者应用程序来异步处理工作负载。从而使得系统能够更有效地处理大量消息和并发请求。
- 可靠性：在异步消息传递下，如果消息传递或处理失败，则可以重新尝试或发送至错误队列以供后续处理，从而提高系统的可靠性。
- 容错能力：异步消息传递通过将消息的生产者和消费者解耦来提供容错能力。如果某个应用程序或组件发生故障，可以存储消息以供系统重新上线后再进行处理，确保数据不会丢失。
缺点
- 复杂性：与其他集成模式相比，要实现异步点对点消息传递模式可能更加复杂，因为需要额外的消息处理逻辑。
- 消息依赖，并且可能产生重复：在异步消息传递系统中，管理消息之间的依赖并确保正确的消息重复恐怕并不容易。这需要精心设计和实现来处理潜在的问题，例如消息顺序、消息重复和消息处理依赖。
- 延迟增加：异步消息传递会导致在发送消息和接收响应之间存在延迟，因为消息的处理时间可能更长。这种延迟会影响实时交互，并且可能不适合需要立即反馈的应用程序。
在做出架构决策时，做出权衡并选择最适合您的特定要求和约束的通信模式始终都很重要。许多现代应用程序依赖于多种集成模式，包括异步点对点消息传递、同步请求响应和基于事件的通信。
不过现在，我们还是先进入教程并学习如何使用 Amazon Lambda 和 Amazon SQS 实现这一模式。
关于编写代码时的资源费用说明：本教程仅使用最少量资源，所有资源都包含在账户创建后的前 12 个月内的亚马逊云科技免费套餐中：
- 大小为几 K 的代码将存储在 Amazon S3 中，Amazon S3 提供 5 GB 的免费存储空间。
- 我们会多次调用 SQS，SQS 提供每月 100 万次的免费请求。
- 我们将在 Amazon Lambda 上调用两个函数，Amazon Lambda 也提供每月 100 万次的免费调用。
因此，如果您遵循本教程逐步指南，您肯定不会超出免费套餐。我还在末尾添加了一个部分，可帮助您删除学习本教程操作的过程中创建的所有资源。
前提条件
如果安装了 CLI 和 CDK，您应该能看到安装的版本。如果尚未安装，您将收到一条错误消息，提示您命令无效。
在开始本教程之前，需要满足以下条件：
- 亚马逊云科技账户：如果尚未拥有账户，您可以免费注册。亚马逊云科技免费套餐为您提供了大量可供使用的资源，包括我们将使用的 Amazon Lambda 和 Amazon SQS。
- Amazon 云开发工具包 (Amazon CDK)：如何设置并引导 Amazon CDK
本教程至少需要版本为 2 的 Amazon CLI 和 Amazon CDK。
您可以通过在终端中执行以下命令来查看 Amazon CLI 版本：
aws --version通过执行以下命令来查看 Amazon CDK 版本：
cdk --version步骤 1 - 创建 CDK 应用程序
1.1 - 初始化 CDK 应用程序
首先，您必须进行初始化设置并让 CDK 自动生成一些用于建立项目的代码。
从命令行为项目创建一个目录，cd 进入该目录，并将项目初始化：
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 - 引导环境
首次将 Amazon CDK 应用程序部署至环境（即特定的亚马逊云科技账户和区域）时，您需要使用 CDKToolkit 堆栈引导环境。该堆栈将部署一个 S3 存储桶，用于在实际应用程序的部署过程中存储模板和资产。
要引导环境，请确保您仍在 point-to-point-example 目录中，然后执行以下命令：
cdk bootstrap这需要一些时间，一旦您在命令行上看到以下输出结果就代表完成：
✅ Environment aws://[account_id]/[region] bootstrapped.注意：如果这一步出现错误，则可能是您尚未 配置 Amazon CLI 或是已经引导了环境。如果是后者，您只需继续下一步即可。
1.3 - 部署初始堆栈
cdk deploy此命令会将应用程序合并至 Amazon CloudFormation 模板中，并将其部署到您的亚马逊云科技账户中。
如果找不到堆栈，请确保点击导航栏中当前显示的区域名称来选择正确的区域，然后选择要切换的区域。确保是您在配置 Amazon CLI 时所使用的堆栈。
现在打开并查看堆栈的详细信息。在 Resources（资源）选项卡中，您可以看到它并未部署任何内容，只有一个名为 CDKMetadata 的资源，这只是我们堆栈的一些基本配置信息。
步骤 2 - 创建 Amazon SQS 队列
一切设置完毕后，您就可以创建第一个真正的资源，就是是一个 SQS 队列。
2.1 - 将队列添加至堆栈中
首先，打开 lib\point-to-point-example-stack.ts 文件。
文件里可能已经添加了一些导入语句，包括注释中的 aws-cdk-lib/aws-sqs，取决于 CDK 模板的版本。在这种情况下，请去掉 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 队列列表并找到名为 SqsTutorialStack-Queue... 的队列：
2.3 - 检查部署好的队列
导航至亚马逊云科技管理控制台中的 SQS 控制面板并查看队列列表。如果找不到队列，请确保您的区域正确。
步骤 3 - 创建生产者函数并将其连接至队列
3.1 - 编写生产者函数源代码
我们将使用 JavaScript 作为 Lambda 函数，并将源代码存储在 CDK 应用程序中的源代码文件夹中。稍后我们将从 CDK 堆栈指向此文件夹，让它自动打包、上传内容并将其部署至亚马逊云科技。
请确保您在项目目录中并创建以下目录和文件：
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;
}这是一个简单的 Lambda 函数处理程序，能将消息发送至我们的 SQS 队列并返回结果。
请注意，由于我们在构建此函数时未必要知道要使用哪个队列，或者可能希望稍后更队列而不必更改函数代码，因此我们从执行环境中检索队列的 URL。这样，我们就可以在 CDK 堆栈中部署 Lambda 函数的过程中进行动态定义。
3.2 - 将生产者函数添加至堆栈中
要在亚马逊云科技中创建真正的 Lambda 函数，您需要将其添加至堆栈中。为此，请打开 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 设置为 SQS 队列的实际 URL，使其可以从函数内部使用 process.env.SQSQueueUrl 进行访问。
3.3 - 向生产者函数授予权限
每个 Lambda 函数都需要一些基本的 IAM 权限，CDK 默认会使用基本执行角色提供这些权限。
但是，我们的生产者函数需要额外的权限才能将消息发送至队列。幸运的是，在 CDK 中，我们不需要手动管理 IAM 策略。我们可以将 SQS 队列的 grantSendMessages 方法结合 Lambda 函数使用。
为此，请在 producerFunction 下面添加下行代码：
queue.grantSendMessages(producerFunction);3.4 - 部署生产者函数
要部署生产者函数及其源代码，请在项目目录中运行以下命令：
cdk deploy这样就为新资源和修改资源的部署做好了准备。
如果对堆栈的更改需要创建或修改 IAM 权限，CDK 会提示您仔细检查更改，确保安全。
在下面的屏幕截图中，您可以看到 CDK 希望部署影响 IAM 权限的三项更改：
- 允许 Lambda 服务（lambda.amazon.aws）承担生产者函数附加的执行角色。
- 直接添加必要的权限，允许执行角色访问 SQS 队列并向其发送消息。
- 将具有基本权限的托管策略添加至执行角色。
键入 y 进行确认，CDK 将部署新的 Lambda 函数及其 IAM 角色。
3.5 - 测试生产者函数
部署函数后，导航至 Amazon Lambda 控制面板 中的函数列表。
函数名称是我们在堆栈中定义的函数名称：Producer（生产者）。如果找不到该函数，请确保您的区域正确。
点击函数名称以打开其详细信息，然后向下滚动至 Code source（代码源）部分。
左侧文件资源管理器中的文件夹包含一个名为 send_message_to_sqs.js 的文件。
双击文件名打开其源代码，您可以看到它正是我们在 lambda-src/producer CDK 项目的文件夹中创建的文件的副本。
现在我们来配置一个测试事件。
为此，请点击 Test（测试）按钮。如果您尚未配置测试事件，就会进入测试配置画面。键入 Event name（事件名），例如“test-event”，然后点击 Save（保存）。
配置好测试事件后，再次点击 Test（测试）便可直接调用 Lambda 函数。
这会创建一个 Execution results（执行结果）选项卡，其中包括以下响应：
{
"$metadata": {
"httpStatusCode": 200,
"requestId": "cb033b2a-1234-5678-9012-66188359d280",
"attempts": 1,
"totalRetryDelay": 0
},
"MD5OfMessageBody": "b5357af0c1c816b2d41275537cc0d1af",
"MessageId": "4a76e538-1234-5678-9012-749f0f4b9294"
}还记得 Lambda 函数中的代码片段吗？
const result = await client.send(command);
return result;上面执行结果中的 JSON 对象正是 SQS 返回的内容，作为对 client.send(command) 的响应，其中包括一些有用的信息，例如 "httpStatusCode": 200，这说明队列已成功接收到消息。
3.6 - 从 SQS 手动检索消息
目前，队列未连接至消费者，因此不会拾取消息。但是，我们可以手动检索消息。
导航至 SQS 控制面板，点击队列以打开其详情页面，然后点击 Send and receive messages（发送和接收消息）按钮。
在下个页面上，向下滚动至 Receive messages（接收消息）部分，然后点击标记为 Poll for messages（轮询消息）的按钮，里面会显示当前正在传输的所有消息，包括我们测试时生产者函数发送的消息：
点击消息 ID 将其打开。里面应该包含以下正文：
{"message":"Hello World from Producer"}恭喜您！您已成功部署并测试了将消息发送至 SQS 队列的 Lambda 函数。
下一步，我们将构建消费者函数并对其进行配置，让其在新消息到达队列时就会被自动调用。
步骤 4 - 创建消费者函数并将其连接至队列
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);
}这是一个非常简单的 Lambda 函数处理程序，用于解包 SQS 消息并将其打印至日志中。
4.2 - 将消费者函数添加至堆栈中
要在亚马逊云科技中创建真正的 Lambda 函数，您需要将其添加至堆栈中。为此，请打开 lib/point-to-point-example-stack.ts。
在构造函数内的 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 队列作为事件源添加至消费者
我们希望只要队列中有新消息就触发消费者函数。为此，我们需要将队列添加为该函数的 Event Source（事件源），并授予该函数使用消息的所有必要权限。
首先，在文件顶部添加另一条 import 语句：
import { SqsEventSource } from 'aws-cdk-lib/aws-lambda-event-sources';接下来，在 consumerFunction 定义下方添加以下两行代码：
consumerFunction.addEventSource(new SqsEventSource(queue));
queue.grantConsumeMessages(consumerFunction);4.4 - 部署消费者函数
要部署生产者函数及其源代码，请在项目目录中运行以下命令：
cdk deploy这样会为部署新资源或修改资源做好准备，并会再次要求您仔细检查新的 IAM 权限。
键入 y 进行确认，CDK 会部署新的 Lambda 函数及其执行角色，并将其连接至 SQS 队列。
部署完成后，一切都将准备就绪并连接完成。我们来测试一下是否生效。
步骤 5 - 应用程序端到端测试
为测试应用程序，我们将手动调用生产者函数以将另一条消息发送至队列。然后我们会检查消费者函数的执行日志，看看消费者是否是自动触发并处理消息。
5.1 - 调用生产者函数
按照 3.5 - 测试生产者函数 中教学步骤再次调用生产者函数。以下是简短总结：
- 导航至 Amazon Lambda 控制面板中的函数列表。
- 点击 Producer（生产者）这个名称来打开生产者函数。
- 向下滚动至 Code source（代码源）部分。
- 点击 Test（测试）按钮。
- 确保函数执行成功。
5.2 - 查看消费者函数的日志
返回至 Amazon Lambda 控制面板中的函数列表，这次打开 Consumer（消费者）函数。
向下滚动页面，打开名为 Monitor（监控）的选项卡，打开里面的 Logs（日志）选项卡，然后查看 Recent Invocations（最近调用），其中您应该能找到函数最近的 LogStream（日志流）链接。
请注意，日志整合至日志流中会需要一些时间。如果没有显示，请等待一分钟左右，然后点击右侧的刷新按钮。
显示出来后，点击日志流链接。里面会显示日志条目，如果您不习惯这种格式，可能看起来会有点混乱。
只需在以 START RequestId: ... 和 END Request Id: ... 开头的两个条目之间查找即可。
点击箭头图标展开条目，您应该能在里面找到原始消息：
大功告成！
恭喜您！您已经创建了使用 SQS 在两个 Lambda 函数之间设置异步点对点消息传递所需的一切资源。
您还可以使用 SQS 执行更多操作，例如批处理消息、确保消息顺序以及自动将错误消息发送至死信队列 (DLQ) 等。要了解更多信息，请参阅 Amazon SQS 开发人员指南。
清理资源
您可以继续尝试使用该应用程序。完成后，您可以通过执行以下命令删除已经部署的所有内容：
cdk destroy在确认步骤后，这会删除作为堆栈一部分而创建的所有资源。
唯一剩下的资源是 Lambda 函数的源代码，源代码也已上传至 Amazon S3 中的资产存储桶中。
如果需要，您可以直接在亚马逊云科技管理控制台的 S3 控制面板 删除该资源：
找到名为 cdk-XXXXXXXXX-assets-YOUR_ACCOUNT_ID-YOUR_REGION 的 S3 存储桶。这是您使用 Amazon CDK 引导账户时创建的存储桶。
您可以通过选择存储桶名称旁边的选中按钮并点击 Empty（清空）来删除存储桶中的所有内容：
这会打开一个确认对话框。仔细检查，确保这是资产存储桶，在文本框中输入 permanently delete（永久删除），然后点击 Empty（清空）。
然后，您还可以删除存储桶。不过，我建议保留存储桶以用于 Amazon CDK 的进一步实验。您无需考虑成本，因为空的 S3 存储桶是免费的。
回顾
在本教程中，我们构建了一个实现异步点对点消息传递模式的简易应用程序。它由两个 Lambda 函数（一个生产者函数和一个消费者函数）组成，这两个函数通过 SQS 队列形成弱耦合。当生产者将消息发送至队列时，消费者会自动拾取消息并通过解包消息并将其打印至日志中来进行处理。
异步点对点消息传递模式是现代 Web 和云架构常用的通信模型，我希望通过本教程的学习，您能够清楚了解如何使用 Amazon SQS 和 Amazon Lambda 实现这一模式。
要了解更多信息，请参阅以下其他资源：
- Amazon Simple Queue Service (SQS) 开发人员指南
- 所有包含 架构模式 标签的文章
- 亚马逊云科技社区中的更多教程
- 更多 ServerlessLand 上的 SQS 消息传递示例
完整源代码
以下是 point-to-point-example/lib/point-to-point-example-stack.ts 中定义的 CDK 堆栈的最终源代码：
import { Construct } from 'constructs';
import { SqsEventSource } from 'aws-cdk-lib/aws-lambda-event-sources';
import * as cdk from 'aws-cdk-lib';
import * as sqs from 'aws-cdk-lib/aws-sqs';
import * as lambda from 'aws-cdk-lib/aws-lambda';
export class PointToPointExampleStack extends cdk.Stack {
constructor(scope: Construct, id: string, props?: cdk.StackProps) {
super(scope, id, props);
// Create the message queue
const queue = new sqs.Queue(this, 'Queue', {
queueName: 'MyQueue',
visibilityTimeout: cdk.Duration.seconds(300),
});
// Create the producer function that will send messages
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,
}
});
// Grant the producer function permission to send messages
queue.grantSendMessages(producerFunction);
// Create the consumer function that will receive messages
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'
});
// Grant the consumer function permission to receive messages
queue.grantConsumeMessages(consumerFunction);
// Add the SQS queue as an event source so that it invokes
// the consumer function whenever there is a new message
consumerFunction.addEventSource(new SqsEventSource(queue));
}
}以下是 point-to-point-example/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;
}以下是 point-to-point-example/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);
}GitHub 存储库
此示例的代码也可通过 GitHub 存储库 获取，其中包括本教程各个步骤的分支。要克隆这些代码，只需执行：
git clone https://github.com/build-on-aws/point-to-point-messaging-with-amazon-sqs.git
