AWS Lambda 事件源

AWS CDK v1在2023-06-01上达到了支持结束. 此软件包不再更新,用户应该迁移到 AWS CDK v2.

关于如何移徙的更多信息,请参见: 移动到 AWS CDK v2 指南。 。 。 .

一个事件源映射是一个AWS Lambda资源,它从事件源读取并引用了Lambda函数. 您可以使用事件源映射来处理不引用 Lambda 服务中的流或队列中的项目 直接函数。 Lambda为以下服务提供事件源映射. 有关羊肉的更多信息 事件源 这里。 。 。 .

此模块包括允许将各种 AWS 服务用作事件的类 aWS Lambda通过高级别 lambda.addEventSource(source) API (英语).

注:在多数情况下,也可以使用资源API来援引一项规则 AWS Lambda 函数 。 这个库为所有 Lambda 事件提供统一的 API 来源,不论它们使用何种基本机制.

以下代码设置了带有 SQS 队列事件源的 lambda 函数 -

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

declare const fn: lambda.Function;
const queue = new sqs.Queue(this, 'MyQueue');
const eventSource = new SqsEventSource(queue);
fn.addEventSource(eventSource);

const eventSourceId = eventSource.eventSourceMappingId;

这个 eventSourceId 属性包含事件源代码。 这将是一个 符号 最终价值的确定 部署.

质量

Amazon 简单队列服务( Amazon SQS) 允许您构建同步 工作流程。 有关亚马逊 SQS 的更多信息,请参见 Amazon 简单队列 服务。 您可以配置 AWS Lambda 来投票这些消息的到来 然后将事件传递给 Lambda 函数引用 。 要查看一个样本事件, 见 亚马逊 SQS 事件。 。 。 .

要建立 Amazon 简单队列服务 作为 AWS Lambda 的事件源, 您 首先创建或更新 Amazon SQS 队列,并选择自定义值 队列参数。 以下参数将影响亚马逊SQS的投票 行为 :

• 可见度 超时: 可能影响复试之间的间隔期.
• 接收Message等时间: 将决定 长 民意调查 持续时间。 默认值为20秒.
• 批量大小: 确定在引用您 lambda 函数前要缓冲多少记录 .
• 最大切入窗口:在引用"羊肉"之前收集记录的最长时间. 这增加了以延误处理为代价进行整批处理的可能性.
• 启用: 如果启用 SQS 事件源映射 。 默认是真实的.

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

const queue = new sqs.Queue(this, 'MyQueue', {
  visibilityTimeout: Duration.seconds(30),      // default,
  receiveMessageWaitTime: Duration.seconds(20), // default
});
declare const fn: lambda.Function;

fn.addEventSource(new SqsEventSource(queue, {
  batchSize: 10, // default
  maxBatchingWindow: Duration.minutes(5),
  reportBatchItemFailures: true, // default to false
}));

第3条

您可以写出 Lambda 函数来处理 S3 桶事件, 如 对象创建或对象删除的事件。 例如,当用户上传一个 照片到一桶,你可能想要亚马逊 S3 引用您的 Lambda 函数,所以 它读取图像,为照片创建缩略图.

您可以使用 Amazon S3 中的桶通知配置功能 配置事件源映射, 识别您想要的桶事件 Amazon S3要发布,而Lambda的功能就是引用.

import * as s3 from '@aws-cdk/aws-s3';
import { S3EventSource } from '@aws-cdk/aws-lambda-event-sources';

const bucket = new s3.Bucket(this, 'mybucket');
declare const fn: lambda.Function;

fn.addEventSource(new S3EventSource(bucket, {
  events: [ s3.EventType.OBJECT_CREATED, s3.EventType.OBJECT_REMOVED ],
  filters: [ { prefix: 'subdir/' } ], // optional
}));

简讯

您可以写入 Lambda 函数处理 Amazon 简单通知服务 通知书。 当向亚马逊 SNS 专题发布消息时, 服务 可以通过传递消息有效载荷作为参数来引用你的Lambda函数. 您的 Lambda 函数代码可以处理事件, 例如发布 消息给其他亚马逊 SNS 主题,或者将消息发送给其他 AWS 服务.

这也使您能够针对亚马逊启动 Lambda 函数 Cloud Watch警报器和其他使用亚马逊SNS的AWS服务.

举例来说,见 附录:信息和JSON 格式 财务报告和审定财务报表 亚马逊 SNS 样本 活动。 。 。 . 例如,请见 使用 AWS Lambda 与 Amazon SNS 来自不同的 账户。 。 。 .

import * as sns from '@aws-cdk/aws-sns';
import { SnsEventSource } from '@aws-cdk/aws-lambda-event-sources';

declare const topic: sns.Topic;
const deadLetterQueue = new sqs.Queue(this, 'deadLetterQueue');

declare const fn: lambda.Function;
fn.addEventSource(new SnsEventSource(topic, {
  filterPolicy: { },
  deadLetterQueue: deadLetterQueue,
}));

当一个用户以您的 Lambda 函数为主题给 SNS 发布 API 打电话时 amazon SNS将调用Lambda来引用您的功能 一个同步。 Lambda将返回交付状态。 如果有错误 amazon SNS将重取Lambda的功能 时间。 三次尝试后,如果亚马逊 SNS 仍然无法成功引用 Lambda 函数,然后 Amazon SNS 将发送发送状态失败消息到 云观.

迪纳摩DB流

您可以从 DynamoDB 表格中写入 Lambda 函数来处理更改事件 。 当写入( 插入、 删除、 更新) 时, 事件会释放到 DynamoDB 流( 如果配置) 操作按表格进行。 见 使用 AWS Lambda 与 Amazon DynataDB 以获取更多关于 Lambda 函数事件源与 DynamoDB 配置的信息.

要处理 Lambda 函数的事件,首先创建或更新 DynamotB 表格,并启用 stream 规格。 然后,创建一个 DynamoEventSource 并添加到您的 Lambda 函数中。 以下参数会影响亚马逊DynamoDB的投票行为: 亚马逊DynamoDB:

• 批量大小: 确定在引用您 lambda 函数之前缓冲了多少记录 - 可能影响您函数的内存用量( 如果太高) 以及跟随来的数据速度( 如果太低) 的能力 .
• 二节节选取错误:如果一批件遇到出错,这会导致批件被一分为二并进行再试,使出错的记录能够被隔离.
• 报告批次项目失败:允许函数返回一批记录的部分成功响应.
• 最大切入窗口:在引用"羊肉"之前收集记录的最长时间. 这增加了以延误处理为代价进行整批处理的可能性.
• 最大记录年龄: 将发送给函数处理的记录的最大年龄 。 超过最大年龄的记录将被视为失败.
• 失败: 如果记录在所有重试后失败, 或记录年龄已超过配置值, 则记录将被发送到此处指定的 SQS 队列或 SNS 主题
• 并行因素:每个硬度上同时处理的批次数.
• 重试尝试:如果失败,记录的最大次数应当重审.
• 启动选项: 将确定消费地点, 要么在最近('LATEST')的记录中, 要么是在最古老的记录中('TRIM HORIZON'). “ TRIM Horizon” 将确保您处理所有可用的数据, 而“ LATEST” 将忽略在附加事件源之前到达的所有记录 .
• 翻转窗口: 使用流时处理窗口的持续时间, 以秒为单位.
• 启用: 如果启用 DynamoDB Streams 事件源映射 。 默认是真实的.

import * as dynamodb from '@aws-cdk/aws-dynamodb';
import { DynamoEventSource, SqsDlq } from '@aws-cdk/aws-lambda-event-sources';

declare const table: dynamodb.Table;

const deadLetterQueue = new sqs.Queue(this, 'deadLetterQueue');

declare const fn: lambda.Function;
fn.addEventSource(new DynamoEventSource(table, {
  startingPosition: lambda.StartingPosition.TRIM_HORIZON,
  batchSize: 5,
  bisectBatchOnError: true,
  onFailure: new SqsDlq(deadLetterQueue),
  retryAttempts: 10,
}));

键盘

您可以写入 Lambda 函数, 处理 Amazon Kinesis Streams 中的流数据 。 关于Amazon Kinesis的更多信息,请参见: 亚马逊 Kinesis 服务。为了更多地了解 Lambda 函数事件源与动能的配置,并查看一个样本事件, 见 亚马逊Kinesis事件。 。 。 .

Amazon Kinesis作为AWS Lambda的事件源,你 首先创建或更新 Amazon Kinesis 流并选择自定义值 事件源参数。 以下参数将影响亚马逊的投票 行为 :

• 批量大小: 确定在引用您 lambda 函数之前缓冲了多少记录 - 可能影响您函数的内存用量( 如果太高) 以及跟随来的数据速度( 如果太低) 的能力 .
• 二节节选取错误:如果一批件遇到出错,这会导致批件被一分为二并进行再试,使出错的记录能够被隔离.
• 报告批次项目失败:允许函数返回一批记录的部分成功响应.
• 最大切入窗口:在引用"羊肉"之前收集记录的最长时间. 这增加了整批货物处理的可能性,而代价可能是延误处理.
• 最大记录年龄: 将发送给函数处理的记录的最大年龄 。 超过最大年龄的记录将被视为失败.
• 失败: 如果记录失败并耗入所有重试, 记录将被发送到此处指定的 SQS 队列或 SNS 主题
• 并行因素:每个硬度上同时处理的批次数.
• 重试尝试:如果失败,记录的最大次数应当重审.
• 启动选项: 将确定消费地点, 要么在最近('LATEST')的记录中, 要么是在最古老的记录中('TRIM HORIZON'). “ TRIM Horizon” 将确保您处理所有可用的数据, 而“ LATEST” 将忽略在附加事件源之前到达的所有记录 .
• 翻转窗口: 使用流时处理窗口的持续时间, 以秒为单位.
• 启用: 如果启用 DynamoDB Streams 事件源映射 。 默认是真实的.

import * as kinesis from '@aws-cdk/aws-kinesis';
import { KinesisEventSource } from '@aws-cdk/aws-lambda-event-sources';

const stream = new kinesis.Stream(this, 'MyStream');

declare const myFunction: lambda.Function;
myFunction.addEventSource(new KinesisEventSource(stream, {
  batchSize: 100, // default
  startingPosition: lambda.StartingPosition.TRIM_HORIZON,
}));

卡夫卡语Name

您可以写入 Lambda 函数来处理来自 亚马逊 MSK 或一个 自我管理 Kafka 分组.

以下代码设置了Amazon MSK,作为Lambda函数的事件源. 需要配置全权证书才能进入 MSK 集群,如 用户名/ 名词认证。 。 。 .

import { Secret } from '@aws-cdk/aws-secretsmanager';
import { ManagedKafkaEventSource } from '@aws-cdk/aws-lambda-event-sources';

// Your MSK cluster arn
const clusterArn = 'arn:aws:kafka:us-east-1:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4';

// The Kafka topic you want to subscribe to
const topic = 'some-cool-topic';

// The secret that allows access to your MSK cluster
// You still have to make sure that it is associated with your cluster as described in the documentation
const secret = new Secret(this, 'Secret', { secretName: 'AmazonMSK_KafkaSecret' });

declare const myFunction: lambda.Function;
myFunction.addEventSource(new ManagedKafkaEventSource({
  clusterArn,
  topic: topic,
  secret: secret,
  batchSize: 100, // default
  startingPosition: lambda.StartingPosition.TRIM_HORIZON,
}));

以下代码设置自控的Kafka集群作为事件源. 基于用户名和密码的认证 将需按 管理访问和许可。 。 。 .

import { Secret } from '@aws-cdk/aws-secretsmanager';
import { SelfManagedKafkaEventSource } from '@aws-cdk/aws-lambda-event-sources';

// The list of Kafka brokers
const bootstrapServers = ['kafka-broker:9092'];

// The Kafka topic you want to subscribe to
const topic = 'some-cool-topic';

// The secret that allows access to your self hosted Kafka cluster
declare const secret: Secret;

declare const myFunction: lambda.Function;
myFunction.addEventSource(new SelfManagedKafkaEventSource({
  bootstrapServers: bootstrapServers,
  topic: topic,
  secret: secret,
  batchSize: 100, // default
  startingPosition: lambda.StartingPosition.TRIM_HORIZON,
}));

如果您自己管理的 Kafka 集群只能通过 VPC 实现, 也可以配置 vpc vpcSubnets 财务报告和审定财务报表 securityGroup。 。 。 .

路线图

最终,本模块将支持下面描述的所有事件源 支持的活动 来源 在 AWS Lambda 开发者指南中.