用于Azure Functions的Durable Functions中的数据持久性和序列化

这个例子需要 Azure。Storage.Blobs NuGet 包。

using System;
using System.Threading.Tasks;
using Azure.Storage.Blobs;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;

public record BlobReference(string Container, string BlobName);

public static class LargePayloadFunctions
{
    [Function("ProcessLargePayload")]
    public static async Task<BlobReference> RunOrchestrator(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        BlobReference inputReference = context.GetInput<BlobReference>()
            ?? throw new InvalidOperationException("A blob reference is required.");

        return await context.CallActivityAsync<BlobReference>(
            nameof(ProcessLargePayloadActivity), inputReference);
    }

    [Function(nameof(ProcessLargePayloadActivity))]
    public static async Task<BlobReference> ProcessLargePayloadActivity(
        [ActivityTrigger] BlobReference inputReference)
    {
        string connectionString =
            Environment.GetEnvironmentVariable("PAYLOAD_STORAGE_CONNECTION_STRING")
            ?? throw new InvalidOperationException("Payload storage is not configured.");
        BlobServiceClient service = new BlobServiceClient(connectionString);

        BlobClient inputBlob = service
            .GetBlobContainerClient(inputReference.Container)
            .GetBlobClient(inputReference.BlobName);
        BinaryData inputData = (await inputBlob.DownloadContentAsync()).Value.Content;

        BlobContainerClient outputContainer =
            service.GetBlobContainerClient("processed-payloads");
        await outputContainer.CreateIfNotExistsAsync();

        string outputName = $"processed/{Guid.NewGuid():N}.json";
        await outputContainer.GetBlobClient(outputName)
            .UploadAsync(inputData, overwrite: true);

        return new BlobReference(outputContainer.Name, outputName);
    }
}

这个例子需要 Azure。Storage.Blobs NuGet 包。

using System;
using System.Threading.Tasks;
using Azure.Storage.Blobs;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;

public class BlobReference
{
    public string Container { get; set; } = string.Empty;
    public string BlobName { get; set; } = string.Empty;
}

public static class LargePayloadFunctions
{
    [FunctionName("ProcessLargePayload")]
    public static async Task<BlobReference> RunOrchestrator(
        [OrchestrationTrigger] IDurableOrchestrationContext context)
    {
        BlobReference inputReference = context.GetInput<BlobReference>()
            ?? throw new InvalidOperationException("A blob reference is required.");

        return await context.CallActivityAsync<BlobReference>(
            "ProcessLargePayloadActivity", inputReference);
    }

    [FunctionName("ProcessLargePayloadActivity")]
    public static async Task<BlobReference> RunActivity(
        [ActivityTrigger] BlobReference inputReference)
    {
        string connectionString =
            Environment.GetEnvironmentVariable("PAYLOAD_STORAGE_CONNECTION_STRING")
            ?? throw new InvalidOperationException("Payload storage is not configured.");
        BlobServiceClient service = new BlobServiceClient(connectionString);

        BlobClient inputBlob = service
            .GetBlobContainerClient(inputReference.Container)
            .GetBlobClient(inputReference.BlobName);
        BinaryData inputData = (await inputBlob.DownloadContentAsync()).Value.Content;

        BlobContainerClient outputContainer =
            service.GetBlobContainerClient("processed-payloads");
        await outputContainer.CreateIfNotExistsAsync();

        string outputName = $"processed/{Guid.NewGuid():N}.json";
        await outputContainer.GetBlobClient(outputName)
            .UploadAsync(inputData, overwrite: true);

        return new BlobReference
        {
            Container = outputContainer.Name,
            BlobName = outputName,
        };
    }
}

这个例子需要 @azure/storage-blob npm 包。

const { randomUUID } = require("crypto");
const { BlobServiceClient } = require("@azure/storage-blob");
const df = require("durable-functions");

df.app.orchestration("processLargePayload", function* (context) {
    const inputReference = context.df.getInput();
    return yield context.df.callActivity(
        "processLargePayloadActivity", inputReference);
});

df.app.activity("processLargePayloadActivity", {
    handler: async (inputReference) => {
        const connectionString =
            process.env.PAYLOAD_STORAGE_CONNECTION_STRING;
        if (!connectionString) {
            throw new Error(
                "PAYLOAD_STORAGE_CONNECTION_STRING is not set.");
        }

        const service = BlobServiceClient.fromConnectionString(
            connectionString);

        const inputBlob = service
            .getContainerClient(inputReference.container)
            .getBlockBlobClient(inputReference.blobName);
        const inputData = await inputBlob.downloadToBuffer();

        const outputContainer =
            service.getContainerClient("processed-payloads");
        await outputContainer.createIfNotExists();

        const outputName = `processed/${randomUUID()}.json`;
        await outputContainer.getBlockBlobClient(outputName)
            .uploadData(inputData);

        return {
            container: outputContainer.containerName,
            blobName: outputName,
        };
    },
});

这个例子需要 azure-storage-blob 包。 加到 requirements.txt。

import os
import uuid

import azure.functions as func
import azure.durable_functions as df
from azure.core.exceptions import ResourceExistsError
from azure.storage.blob import BlobServiceClient

app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)


@app.orchestration_trigger(context_name="context")
def process_large_payload(context: df.DurableOrchestrationContext):
    input_reference: dict = context.get_input()
    return (yield context.call_activity(
        "process_large_payload_activity", input_reference))


@app.activity_trigger(input_name="input_reference")
def process_large_payload_activity(input_reference: dict) -> dict:
    connection_string = os.getenv("PAYLOAD_STORAGE_CONNECTION_STRING")
    if not connection_string:
        raise ValueError(
            "PAYLOAD_STORAGE_CONNECTION_STRING is not set.")

    service = BlobServiceClient.from_connection_string(connection_string)

    input_blob = service.get_blob_client(
        container=input_reference["container"],
        blob=input_reference["blobName"])
    input_data = input_blob.download_blob().readall()

    output_container = service.get_container_client("processed-payloads")
    try:
        output_container.create_container()
    except ResourceExistsError:
        pass

    output_name = f"processed/{uuid.uuid4()}.json"
    output_container.upload_blob(
        name=output_name, data=input_data, overwrite=True)

    return {
        "container": output_container.container_name,
        "blobName": output_name,
    }

本示例需要 Az.Storage 模块。 把它添加到 requirements.psd1 里，并确保在 host.json中启用了托管依赖。

编曲者 run.ps1：

param($Context)

$OutputReference = Invoke-DurableActivity `
    -FunctionName 'ProcessLargePayloadActivity' `
    -Input $Context.Input

return $OutputReference

活动：run.ps1

param($InputReference)

if ([string]::IsNullOrEmpty(
        $env:PAYLOAD_STORAGE_CONNECTION_STRING)) {
    throw 'The PAYLOAD_STORAGE_CONNECTION_STRING app setting is not configured.'
}

$StorageContext = New-AzStorageContext `
    -ConnectionString $env:PAYLOAD_STORAGE_CONNECTION_STRING
$InputFile = New-TemporaryFile

try {
    Get-AzStorageBlobContent `
        -Container $InputReference.container `
        -Blob $InputReference.blobName `
        -Destination $InputFile.FullName `
        -Context $StorageContext `
        -Force | Out-Null

    # Process $InputFile here. This example uploads it unchanged.
    $OutputContainer = 'processed-payloads'
    New-AzStorageContainer `
        -Name $OutputContainer `
        -Context $StorageContext `
        -ErrorAction SilentlyContinue | Out-Null

    $OutputName = "processed/$([guid]::NewGuid()).json"
    Set-AzStorageBlobContent `
        -File $InputFile.FullName `
        -Container $OutputContainer `
        -Blob $OutputName `
        -Context $StorageContext `
        -Force | Out-Null

    return @{
        container = $OutputContainer
        blobName = $OutputName
    }
}
finally {
    Remove-Item $InputFile.FullName -Force
}

使用orchestrationTrigger中描述的标准 activityTrigger 和 绑定。

此示例需要 com.azure:azure-storage-blob 和 com.microsoft:durabletask-azure-functions Maven 软件包。

import com.azure.core.util.BinaryData;
import com.azure.storage.blob.BlobContainerClient;
import com.azure.storage.blob.BlobServiceClient;
import com.azure.storage.blob.BlobServiceClientBuilder;
import com.microsoft.azure.functions.annotation.FunctionName;
import com.microsoft.durabletask.TaskOrchestrationContext;
import com.microsoft.durabletask.azurefunctions.DurableActivityTrigger;
import com.microsoft.durabletask.azurefunctions.DurableOrchestrationTrigger;

import java.util.Objects;
import java.util.UUID;

public class LargePayloadFunctions {
    public static class BlobReference {
        public String container;
        public String blobName;

        public BlobReference() {
        }

        public BlobReference(String container, String blobName) {
            this.container = container;
            this.blobName = blobName;
        }
    }

    @FunctionName("ProcessLargePayload")
    public BlobReference runOrchestrator(
            @DurableOrchestrationTrigger(name = "ctx")
            TaskOrchestrationContext ctx) {
        BlobReference inputReference = Objects.requireNonNull(
            ctx.getInput(BlobReference.class),
            "A blob reference is required.");
        return ctx.callActivity(
            "ProcessLargePayloadActivity",
            inputReference,
            BlobReference.class).await();
    }

    @FunctionName("ProcessLargePayloadActivity")
    public BlobReference runActivity(
            @DurableActivityTrigger(name = "inputReference")
            BlobReference inputReference) {
        String connectionString = Objects.requireNonNull(
            System.getenv("PAYLOAD_STORAGE_CONNECTION_STRING"),
            "PAYLOAD_STORAGE_CONNECTION_STRING is not configured.");
        BlobServiceClient service = new BlobServiceClientBuilder()
            .connectionString(connectionString)
            .buildClient();

        BinaryData inputData = service
            .getBlobContainerClient(inputReference.container)
            .getBlobClient(inputReference.blobName)
            .downloadContent();

        BlobContainerClient outputContainer =
            service.getBlobContainerClient("processed-payloads");
        outputContainer.createIfNotExists();

        String outputName =
            "processed/" + UUID.randomUUID() + ".json";
        outputContainer.getBlobClient(outputName)
            .upload(inputData, true);

        return new BlobReference(
            outputContainer.getBlobContainerName(), outputName);
    }
}

.NET 隔离和 System.Text.Json

在 .NET 隔离工作进程 中运行的 Durable Functions 使用为 Azure Functions 应用全局配置的同一个对象序列化器（请参见 WorkerOptions）。 该串行器默认是 System.Text.Json ，而非 Newtonsoft.Json。 对 WorkerOptions.Serializer 的任何更改都会传导到 Durable Functions。

有关 .NET 中 JSON 序列化的内置支持的详细信息，请参阅 .NET 中的 JSON 序列化和反序列化概述文档。

默认序列化逻辑

适用于 .NET（进程内）的 Durable Functions 在内部使用 Json.NET，以便将业务流程和实体数据序列化为 JSON。 使用的默认 Json.NET 设置是：

输入、输出和状态：

JsonSerializerSettings
{
    TypeNameHandling = TypeNameHandling.None,
    DateParseHandling = DateParseHandling.None,
}

异常：

JsonSerializerSettings
{
    ContractResolver = new ExceptionResolver(),
    TypeNameHandling = TypeNameHandling.Objects,
    ReferenceLoopHandling = ReferenceLoopHandling.Ignore,
}

在JsonSerializerSettings阅读有关 的详细文档。

使用.NET属性自定义序列化

在序列化期间，Json.NET 查找类和属性上的各种属性，这些属性控制数据如何从 JSON 序列化和反序列化。 如果你拥有传递到 Durable Functions API 的数据类型的源代码，请考虑将这些特性添加到类型以自定义序列化和反序列化。

通过依赖注入自定义序列化

面向 .NET 并在 Functions V3 运行时上运行的函数应用可以使用依赖项注入 (DI) 来自定义数据和异常的序列化方式。 以下示例代码演示如何使用 DI 通过 IMessageSerializerSettingsFactory 和 IErrorSerializerSettingsFactory 服务接口的自定义实现替代默认 Json.NET序列化设置。

using Microsoft.Azure.Functions.Extensions.DependencyInjection;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Extensions.DependencyInjection;
using Newtonsoft.Json;
using System.Collections.Generic;

[assembly: FunctionsStartup(typeof(MyApplication.Startup))]
namespace MyApplication
{
    public class Startup : FunctionsStartup
    {
        public override void Configure(IFunctionsHostBuilder builder)
        {
            builder.Services.AddSingleton<IMessageSerializerSettingsFactory, CustomMessageSerializerSettingsFactory>();
            builder.Services.AddSingleton<IErrorSerializerSettingsFactory, CustomErrorSerializerSettingsFactory>();
        }

        /// <summary>
        /// A factory that provides the serialization for all inputs and outputs for activities and
        /// orchestrations, as well as entity state.
        /// </summary>
        internal class CustomMessageSerializerSettingsFactory : IMessageSerializerSettingsFactory
        {
            public JsonSerializerSettings CreateJsonSerializerSettings()
            {
                // Return your custom JsonSerializerSettings here
            }
        }

        /// <summary>
        /// A factory that provides the serialization for all exceptions thrown by activities
        /// and orchestrations
        /// </summary>
        internal class CustomErrorSerializerSettingsFactory : IErrorSerializerSettingsFactory
        {
            public JsonSerializerSettings CreateJsonSerializerSettings()
            {
                // Return your custom JsonSerializerSettings here
            }
        }
    }
}

序列化和反序列化逻辑

Azure Functions Node 应用程序使用 JSON.stringify() 进行序列化，使用 JSON.Parse() 进行反序列化。 大多数类型都会无缝地进行序列化和反序列化。 如果默认逻辑不够用，则可以通过在对象上定义一个toJSON()方法来覆盖序列化逻辑。 但是，对于对象反序列化，不存在模拟。

若要对序列化/反序列化管道进行完全的自定义，请考虑使用你自己的代码来处理序列化和反序列化，并将数据作为字符串进行传递。

序列化和反序列化逻辑

Durable Functions PowerShell 应用使用默认的 PowerShell JSON 处理方式来处理业务流程和活动的有效负载。 在大多数情况下，兼容 JSON 的普通对象能够正确序列化和反序列化，无需额外配置。

如果你需要为负载形状自定义行为，可以在函数内自己序列化或反序列化数据，并在编排器、活动和实体之间传递兼容 JSON 的值。 这种方法保持数据格式的明确性，避免依赖未为PowerShell暴露的串行器定制API。

对于更复杂的场景，可以考虑将数据表示为简单的字符串、数组或哈希表，并在函数代码中转换不同格式。

序列化和反序列化逻辑

建议使用类型注解以确保 Durable Functions 正确地序列化和反序列化数据。 虽然许多内置类型是自动处理的，但是某些内置数据类型需要类型注解才能在反序列化期间保留类型。

对于自定义数据类型，可以通过在数据类型类中定义类方法to_json和from_json来实现 JSON 序列化和反序列化。 请注意，这些方法不会在业务流程协调程序函数的返回值上调用，这意味着返回值必须是本机 JSON 可序列化的。 有关详细信息，请参阅 绑定。

要验证反序列化的负载是否符合预期类型，并可选选择加入硬化严格模式，请参见 Python 的 Durable Functions 中的“迁移到类型安全序列化”。

Java 使用 Jackson v2.x 库对数据有效负载进行序列化和反序列化。 可以在 POJO 类型上使用 Jackson 注释来自定义序列化行为。