如何繁殖
在这个答案的末尾,有一个完整的、最小的例子,可以很容易地重现这个问题。
要部署,请创建所有文件并填写aws
profile
并期望
region
进入所有
sh
文件夹。
然后跑
.deploy-stack.sh
创建包含所有必要资源的cloudformation堆栈。
然后打开AWS web界面(SQS)并运行
生成消息.sh
在队列中生成消息。
然后可以看到,在函数节流和队列完全停止之前,大约有一半的消息被处理。
在所有调试完成后删除cloudformation堆栈
remove-stack.sh
解决方案
AWS文档
contains a note
说
为了让您的函数有时间处理每一批记录,请将源队列的可见性超时设置为您在函数上配置的超时的至少6倍。如果您的函数在处理前一批时执行受阻,额外的时间允许Lambda重试。
更改
timeout
关于lambda函数
600
到
100
重新部署堆栈允许所有消息正确处理,即使lambda函数节流。
我无法解释为什么会观察到这种行为,非常感谢对此的反馈。然而,上述内容确实解决了所描述的问题。
文件夹
stack.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Debug Stack for Fifo with Lambda Processor
Resources:
MyLambdaRole:
Type: AWS::IAM::Role
Properties:
RoleName:
Fn::Sub: lambda-role
AssumeRolePolicyDocument:
Version: 2012-10-17
Statement:
- Action:
- sts:AssumeRole
Effect: Allow
Principal:
Service:
- lambda.amazonaws.com
ManagedPolicyArns:
- arn:aws:iam::aws:policy/AWSLambdaExecute
- arn:aws:iam::aws:policy/AmazonSqsFullAccess
Path: /
MySqsQueue:
Type: 'AWS::SQS::Queue'
Properties:
FifoQueue: true
VisibilityTimeout: 600
MySQSQueueFunction:
Type: AWS::Lambda::Function
Properties:
Handler: index.handler
Role: !GetAtt MyLambdaRole.Arn
Runtime: nodejs12.x
Timeout: 600
ReservedConcurrentExecutions: 5
Code:
ZipFile: |
exports.handler = (event, context) => new Promise((resolve) => {
setTimeout(resolve, 1000);
});
MySQSLambdaEventSource:
Type: AWS::Lambda::EventSourceMapping
Properties:
BatchSize: 1
Enabled: false
EventSourceArn: !GetAtt MySqsQueue.Arn
FunctionName: !Ref MySQSQueueFunction
Outputs:
QueueUrl:
Value:
Ref: MySqsQueue
EventSource:
Value:
Ref: MySQSLambdaEventSource
deploy-stack.sh
#!/bin/bash
profile=local
region=us-east-1
# -----------------
aws cloudformation deploy \
--profile $profile \
--region $region \
--template-file stack.yaml \
--stack-name fifo-lambda-debug \
--capabilities CAPABILITY_NAMED_IAM
generate-messages.sh
#!/bin/bash
profile=local
region=us-east-1
# -----------------
function genGroupId {
echo $(shuf -i 1-10 -n 1)
}
function genRndStr {
echo $(openssl rand -hex 12)
}
function entry {
echo "{\"Id\":\"$(genRndStr)\",\"MessageBody\":\"$(genRndStr)\",\"MessageGroupId\":\"$(genGroupId)\",\"MessageDeduplicationId\":\"$(genRndStr)\"}"
}
# -----------------
echo "Getting Subscription UUID..."
eventSource=$(aws cloudformation describe-stacks \
--query "Stacks[0].Outputs[?OutputKey=='EventSource'].OutputValue" \
--output text \
--profile $profile \
--region $region \
--stack-name fifo-lambda-debug)
echo "Getting Queue Url..."
queueUrl=$(aws cloudformation describe-stacks \
--query "Stacks[0].Outputs[?OutputKey=='QueueUrl'].OutputValue" \
--output text \
--profile $profile \
--region $region \
--stack-name fifo-lambda-debug)
echo "Disabling Subscription"
aws lambda update-event-source-mapping \
--profile $profile \
--region $region \
--uuid $eventSource \
--no-enabled \
> /dev/null
while : ; do
echo "Waiting until Subscription disabled..."
[[ $(aws lambda get-event-source-mapping \
--profile $profile \
--region $region \
--uuid $eventSource \
--query "State") != '"Disabled"' ]] || break
sleep 10
done
echo "Queueing Messages..."
for i in {1..30}
do
aws sqs send-message-batch \
--profile $profile \
--region $region \
--queue-url "$queueUrl" \
--entries "[$(entry),$(entry),$(entry),$(entry),$(entry),$(entry),$(entry),$(entry),$(entry),$(entry)]" \
> /dev/null
echo "Done: $i / 30"
done
echo "Re-Enabling Subscription..."
aws lambda update-event-source-mapping \
--profile $profile \
--region $region \
--uuid $eventSource \
--enabled \
> /dev/null
remove-stack.sh
#!/bin/bash
profile=local
region=us-east-1
# -----------------
aws cloudformation delete-stack \
--profile $profile \
--region $region \
--stack-name fifo-lambda-debug
