如何修改Azure Durable Functions代码测算函数间数据传输耗时？

实现逻辑 #

按照需求，我们将在Activity函数中生成指定数据、记录发送前时间，把数据和时间戳打包返回给Orchestrator；Orchestrator接收后记录接收时间，最终计算两者的时间差得到传输耗时。

完整代码实现 #

1. Orchestrator 函数 #

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

def orchestrator_function(context: df.DurableOrchestrationContext):
    # 配置数据参数：大小（字节）、类型
    data_config = {
        "size": 1024 * 1024,  # 1MB
        "type": "string"
    }

    # 调用Activity函数，获取数据和发送时间戳
    activity_result = yield context.call_activity("GenerateAndSendData", data_config)
    send_timestamp = activity_result["send_time"]

    # 记录接收完成时间
    receive_timestamp = datetime.now().timestamp()

    # 计算传输耗时（转换为毫秒）
    transfer_duration_ms = (receive_timestamp - send_timestamp) * 1000

    # 输出结果（可替换为日志存储、数据库写入等操作）
    print(f"数据传输耗时: {transfer_duration_ms:.2f} 毫秒")
    print(f"数据大小: {data_config['size']} 字节")

    return {
        "耗时(ms)": round(transfer_duration_ms, 2),
        "数据大小(字节)": data_config["size"],
        "数据类型": data_config["type"]
    }

main = df.Orchestrator.create(orchestrator_function)

2. Activity 函数 #

import azure.functions as func
from datetime import datetime
import random
import string

def main(data_config: dict) -> dict:
    data_size = data_config["size"]
    data_type = data_config["type"]
    generated_data = None

    # 根据类型生成对应大小的数据
    if data_type == "string":
        # 生成随机字符串（单字符约1字节）
        generated_data = ''.join(random.choices(string.ascii_letters + string.digits, k=data_size))
    elif data_type == "bytes":
        # 生成随机字节数据
        generated_data = bytes(random.getrandbits(8) for _ in range(data_size))
    # 可扩展JSON、结构化对象等其他数据类型

    # 记录发送前的时间戳
    send_timestamp = datetime.now().timestamp()

    # 返回数据和时间戳给Orchestrator
    return {
        "data": generated_data,
        "send_time": send_timestamp
    }

3. HTTP 触发函数（启动Orchestrator） #

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

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    instance_id = await client.start_new(req.route_params["functionName"], None, None)

    return client.create_check_status_response(req, instance_id)

关键说明 #

• 时间精度：使用timestamp()获取秒级时间戳，转换为毫秒后保留两位小数，满足耗时测量需求。
• 数据传输：Durable Functions内部会自动处理Activity返回值的序列化、传输和反序列化，代码中记录的耗时包含这一完整流程。
• 扩展性：可根据需求添加更多数据类型生成逻辑，或把耗时结果写入Azure Monitor、Blob Storage等服务做后续分析。

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