如何从Lambda函数实现动态延迟调用另一个Lambda函数？

方案1：使用SQS延迟队列（推荐，简单高效） #

SQS标准队列支持为单条消息设置自定义延迟时长（最大15分钟），完全满足动态配置延迟的需求，且成本极低。

实现步骤： #

• 创建一个SQS标准队列，无需开启队列级别的固定延迟
• 配置该SQS队列作为Lambda函数B的触发源（在Lambda控制台的触发器中添加SQS触发器）
• 在Lambda函数A中，当收到"success"消息时，计算需要的延迟秒数，调用SQS的send_message接口，指定DelaySeconds参数为计算出的时长

代码示例（Python）： #

import boto3
import os

sqs = boto3.client('sqs')
QUEUE_URL = os.environ['SQS_QUEUE_URL']

def lambda_handler(event, context):
    if "success" in event.get('message', ''):
        # 动态设置延迟时长，此处为60秒
        delay_seconds = 60
        sqs.send_message(
            QueueUrl=QUEUE_URL,
            MessageBody='trigger lambda B',
            DelaySeconds=delay_seconds
        )
    return {"status": "ok"}

注意：若需超过15分钟的延迟，可结合SQS死信队列+EventBridge循环处理，你的场景（1分钟延迟）直接使用上述方案即可。

方案2：使用EventBridge动态定时触发 #

EventBridge支持发送带有指定触发时间的事件，可在Lambda A中计算延迟后的目标时间，调用EventBridge的put_events接口指定触发时间，EventBridge会在该时间点触发Lambda B。

实现步骤： #

• 为Lambda B添加EventBridge触发器，允许EventBridge调用该函数
• 在Lambda函数A中，计算延迟后的时间（当前时间+延迟时长），构造EventBridge事件并发送

代码示例（Python）： #

import boto3
import os
from datetime import datetime, timedelta
import json

eventbridge = boto3.client('events')
TARGET_ARN = os.environ['LAMBDA_B_ARN']

def lambda_handler(event, context):
    if "success" in event.get('message', ''):
        # 动态计算1分钟后的触发时间
        target_time = datetime.utcnow() + timedelta(minutes=1)
        eventbridge.put_events(
            Entries=[
                {
                    'Source': 'lambda.a.trigger',
                    'DetailType': 'invoke lambda b',
                    'Detail': json.dumps({"trigger_from": "lambda_a"}),
                    'Time': target_time,
                    'EventBusName': 'default',
                    'Targets': [
                        {
                            'Arn': TARGET_ARN,
                            'Id': 'lambda-b-target'
                        }
                    ]
                }
            ]
        )
    return {"status": "ok"}

该方案支持最长1年的延迟，适合需要超长延迟的场景。

方案3：Step Functions动态等待（适配复杂工作流） #

你提到的Step Functions并非只能配置固定延迟，它支持通过输入参数传递动态等待时长，使用Wait状态的SecondsPath或TimestampPath引用动态值即可。

实现步骤： #

• 创建Step Functions状态机，包含一个Wait状态（用SecondsPath引用输入中的延迟秒数），后续连接调用Lambda B的Task状态
• 在Lambda函数A中，收到"success"消息时启动状态机，并传入动态延迟秒数参数

状态机定义示例（JSON）： #

{
  "Comment": "Dynamic delay to invoke Lambda B",
  "StartAt": "WaitForDynamicDuration",
  "States": {
    "WaitForDynamicDuration": {
      "Type": "Wait",
      "SecondsPath": "$.delay_seconds",
      "Next": "InvokeLambdaB"
    },
    "InvokeLambdaB": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:REGION:ACCOUNT_ID:function:LambdaB",
      "End": true
    }
  }
}

Lambda A调用状态机的代码示例（Python）： #

import boto3
import os
import json

stepfunctions = boto3.client('stepfunctions')
STATE_MACHINE_ARN = os.environ['STATE_MACHINE_ARN']

def lambda_handler(event, context):
    if "success" in event.get('message', ''):
        # 动态传入延迟秒数
        input_params = {"delay_seconds": 60}
        stepfunctions.start_execution(
            stateMachineArn=STATE_MACHINE_ARN,
            input=json.dumps(input_params)
        )
    return {"status": "ok"}

此方案适合存在复杂工作流需求的场景，比如延迟后需后续处理、错误重试等。

内容的提问来源于stack exchange，提问作者yuni lee