python - 异步运行 Azure Durable Function

Question

我正在尝试在 Azure 中运行异步函数，据我所知，这是使用 Azure 中的持久函数完成的。我一直在与tutorial documentation合作来自 Microsoft，但它似乎实际上并不是异步运行的。

这是构造函数演示

import azure.functions as func
import azure.durable_functions as df


def orchestrator_function(context: df.DurableOrchestrationContext):
    result1 = yield context.call_activity('E1_SayHello', "Tokyo")
    result2 = yield context.call_activity('E1_SayHello', "Seattle")
    result3 = yield context.call_activity('E1_SayHello', "London")
    return [result1, result2, result3]

main = df.Orchestrator.create(orchestrator_function)

这是被调用的函数

def main(name: str) -> str:
    logging.info(f"Hello {name}!")
    return f"Hello {name}!"

效果很好，您很快就会收到“Hello {City}!”的三个响应。如您所料。但它似乎一次一个地调用它，在中间等待响应。如果您稍微更改该函数并在其中添加一个 sleep 函数，如下所示，每个响应相隔 10 秒，所需的结果是它们将同时被调用，然后等待所有 3 个响应的结果。

def main(name: str) -> str:
    time.sleep(10)
    logging.info(f"Hello {name}!")
    return f"Hello {name}!"

我尝试用 context.wait_for_external_event 替换 context.call_activity ，但这有类似的结果，并且文档的许多部分仅使用 C# 和 JavaScript，所以我不知道是否没有 Python 或的等效项如果它没有完整记录。

最终目标是能够使用列表中保存的各种参数进行多次 API 调用(60+)，在写入数据库之前清理数据。目前，每个 API 调用只需不到 60 秒，因此希望以异步模式运行它们，这样就不会花费近一个小时。

Answer 1

我最终通过他们的Github sample code找到了答案。所以我会把它放在这里以防其他人正在搜索。

您需要首先将 context.call_activity 函数放入列表中，然后使用该列表作为yield context.task_all 中的函数变量，如下所示。

def orchestrator_function(context: df.DurableOrchestrationContext):
    tasks = []
    cities = ["Tokyo", "Seattle", "London"]
    for city in cities:
        tasks.append(context.call_activity('Hello', city))

    results = yield context.task_all(tasks)
    return results