Azure Durable Functions Python：Blob操作类复用连接的实现疑问

核心结论 #

完全不需要在每个Blob I/O Activity函数中重新创建连接，也不需要把类当作函数使用。你可以通过静态单例类或者依赖注入来统一管理Blob服务客户端，实现连接复用，同时符合Durable Functions的确定性要求。

具体实现方案 #

方案1：静态单例工具类 #

利用BlobServiceClient的线程安全特性，创建一个静态类维护其单例实例，封装所有Blob操作方法，供Activity函数直接调用：

public static class BlobOperationHelper
{
    // 懒加载创建BlobServiceClient单例
    private static readonly Lazy<BlobServiceClient> _blobServiceClient = new Lazy<BlobServiceClient>(() =>
    {
        string connectionString = Environment.GetEnvironmentVariable("AzureWebJobsStorage") 
            ?? throw new InvalidOperationException("未配置Blob连接字符串");
        return new BlobServiceClient(connectionString);
    });

    public static BlobServiceClient Client => _blobServiceClient.Value;

    // 封装Blob列表查询方法
    public static async Task<List<string>> ListBlobsAsync(string containerName)
    {
        var containerClient = Client.GetBlobContainerClient(containerName);
        await containerClient.CreateIfNotExistsAsync();

        var blobNames = new List<string>();
        await foreach (var blob in containerClient.GetBlobsAsync())
        {
            blobNames.Add(blob.Name);
        }
        return blobNames;
    }

    // 封装Blob上传方法
    public static async Task UploadBlobAsync(string containerName, string blobName, Stream content)
    {
        var containerClient = Client.GetBlobContainerClient(containerName);
        await containerClient.CreateIfNotExistsAsync();
        
        var blobClient = containerClient.GetBlobClient(blobName);
        await blobClient.UploadAsync(content, overwrite: true);
    }

    // 按需添加下载等其他操作方法
}

在Activity函数中调用示例：

[FunctionName("ListBlobsActivity")]
public static async Task<List<string>> RunListBlobsActivity(
    [ActivityTrigger] string containerName,
    ILogger log)
{
    return await BlobOperationHelper.ListBlobsAsync(containerName);
}

方案2：依赖注入（推荐） #

如果你的Function App使用隔离进程模型或支持依赖注入，推荐通过DI注册BlobServiceClient为单例，再注入到Activity函数中，这种方式更利于测试和代码维护：

1. 在Program.cs中注册服务：

var host = new HostBuilder()
    .ConfigureFunctionsWorkerDefaults()
    .ConfigureServices(services =>
    {
        services.AddSingleton(_ =>
        {
            string connectionString = Environment.GetEnvironmentVariable("AzureWebJobsStorage")
                ?? throw new InvalidOperationException("未配置Blob连接字符串");
            return new BlobServiceClient(connectionString);
        });
    })
    .Build();

host.Run();

2. 在Activity函数中注入并使用：

public class BlobActivities
{
    private readonly BlobServiceClient _blobServiceClient;

    // 构造函数注入Blob服务客户端
    public BlobActivities(BlobServiceClient blobServiceClient)
    {
        _blobServiceClient = blobServiceClient;
    }

    [FunctionName("UploadBlobActivity")]
    public async Task RunUploadActivity(
        [ActivityTrigger] (string ContainerName, string BlobName, Stream Content) input,
        ILogger log)
    {
        var containerClient = _blobServiceClient.GetBlobContainerClient(input.ContainerName);
        await containerClient.CreateIfNotExistsAsync();
        
        var blobClient = containerClient.GetBlobClient(input.BlobName);
        await blobClient.UploadAsync(input.Content, overwrite: true);
    }
}

关键注意事项 #

• Orchestrator函数限制：无论采用哪种方案，Orchestrator都不能直接调用Blob操作方法，必须通过Activity函数执行所有I/O操作，确保Orchestrator的确定性。
• 连接池优化：BlobServiceClient内部已实现连接池管理，复用单例实例可避免重复建立TCP连接的开销，提升函数运行性能。
• 配置安全性：Blob连接字符串必须从Function App的应用设置中读取，绝对不能硬编码在代码内。

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