使用boto调用lambda函数，如何异步执行？

Question

我正在使用boto来调用我的Lambda函数并测试后端。我希望以异步方式调用它们。我注意到"invoke_async"已被弃用，不应再使用。相反，您应该使用InvocationType为"Event"的"invoke"来异步执行函数。

但是，当函数返回时，我似乎无法得到响应。我尝试了以下方法：

payload3=b"""{
"latitude": 39.5732160891,
"longitude": -119.672918997,
"radius": 100
}"""

client = boto3.client('lambda')
for x in range (0, 5):
    response = client.invoke(
        FunctionName="loadSpotsAroundPoint",
        InvocationType='Event',
        Payload=payload3
    )
    time.sleep(15)
    print(json.loads(response['Payload'].read()))
    print("\n")

即使我告诉代码要睡15秒钟，当我尝试打印它时，响应变量仍然为空。如果我将 InvokationType 更改为 "RequestResponse"，一切都可以正常工作，响应变量会打印出来，但这是同步的。我错过了什么简单的东西吗？当异步调用返回时，如何执行一些代码，例如打印结果？

谢谢。

- sometimesiwritecode

Answer 1

在'async AWS lambda invocation'和'async python code'之间存在差异。当将InvocationType设置为'Event'时，根据定义，它不会发送任何响应。

在您的示例中，invoke()立即返回None，并且没有隐式地启动任何后台程序来更改稍后的这个值 (感谢上帝!)。因此，当您15秒后查看response的值时，它仍然是 None。

看起来您真正想要的是使用异步Python代码的RequestResponse调用类型。您有很多选择，但我最喜欢的是concurrent.futures。另一个选择是threading。

这里是使用concurrent.futures的示例：

(如果您正在使用Python2，则需要pip install futures)

from concurrent.futures import ThreadPoolExecutor
import json

payload = {...}

with ThreadPoolExecutor(max_workers=5) as executor:
    futs = []
    for x in xrange(0, 5):
        futs.append(
            executor.submit(client.invoke,
                FunctionName   = "loadSpotsAroundPoint",
                InvocationType = "RequestResponse",
                Payload        = bytes(json.dumps(payload))
            )
        )
    results = [ fut.result() for fut in futs ]

print results

你可能想要了解的另一种模式是使用事件调用类型，并使您的Lambda函数向SNS推送消息，然后由另一个Lambda函数消耗这些消息。 您可以在此处查看有关SNS触发的lambda函数的教程。

- Julien

---

作为后续说明，我不得不使用bytes(json.dumps(payload), "utf-8")，否则会抛出一个错误。 - Ben Balentine

我不得不使用 Payload = json.dumps(payload)。 - Aslan

@MohammadSadoughi 上面的ThreadPoolExecutor在单个处理器的多个线程上执行此操作。你可能在考虑ProcessPoolExecutor，它启动多个进程（因此可以利用多个处理器）。 - Julien

InvocationType = 'Event' 立即返回。 - grantr

Answer 2

异步执行的AWS Lambda函数不会返回执行结果。如果异步调用请求成功（即没有由于权限等原因导致错误），AWS Lambda会立即返回HTTP状态码202 ACCEPTED，并且不再承担有关此异步调用结果的任何信息通信方面的责任。

从AWS Lambda Invoke action文档中得知：

响应语法

HTTP/1.1 StatusCode
X-Amz-Function-Error: FunctionError
X-Amz-Log-Result: LogResult

Payload

响应元素

如果操作成功，服务将返回以下HTTP响应。

状态码

对于成功的请求，HTTP状态码将在200范围内。 对于RequestResponse调用类型，此状态代码将是200。 对于Event调用类型，此状态代码将是202。 对于DryRun调用类型，状态代码将为204。

[...]

响应返回以下内容作为HTTP正文。

有效负载

它是Lambda函数返回的对象的JSON表示形式。 只有在调用类型为RequestResponse时才存在。

- Leon

Answer 3

以下是一个Python函数，它接受lambda函数名称和要发送到该函数的负载。它通过boto3客户端调用Lambda函数。

import boto3, json, typing

def invokeLambdaFunction(*, functionName:str=None, payload:typing.Mapping[str, str]=None):
    if functionName == None:
        raise Exception('ERROR: functionName parameter cannot be NULL')
    payloadStr = json.dumps(payload)
    payloadBytesArr = bytes(payloadStr, encoding='utf8')
    client = boto3.client('lambda')
    return client.invoke(
        FunctionName=functionName,
        InvocationType="RequestResponse",
        Payload=payloadBytesArr
    )

使用方法：

if __name__ == '__main__':
    payloadObj = {"something" : "1111111-222222-333333-bba8-1111111"}
    response = invokeLambdaFunction(functionName='myLambdaFuncName',  payload=payloadObj)
    print(f'response:{response}')

- ylev

---

获取输出结果的代码：response["Payload"].read() - Max Reeder