# Azure Durable Functions（.NET 隔离）与 Durable Task Scheduler 使用连接到 Azure Durable Task Scheduler 的 Azure Durable Functions 构建容错、有状态的无服务器工作流。 ## 快速入门 ### 所需的 NuGet 包 <项目组> ### host.json 配置（持久任务计划程序） { “版本”：“2.0”， “扩展”：{ “持久任务”：{ “存储提供者”：{ “类型”：“azureManaged”， “连接字符串名称”：“DTS_CONNECTION_STRING” }, "hubName": "%TASKHUB_NAME%" } } } ### local.settings.json { “IsEncrypted”：假， “价值观”：{ "FUNCTIONS_WORKER_RUNTIME": "dotnet 隔离", "AzureWebJobsStorage": "UseDevelopmentStorage=true", "DTS_CONNECTION_STRING": "端点=http://localhost:8080;身份验证=无", "TASKHUB_NAME": "默认" } } ### 最小示例（基于函数）使用 Microsoft.Azure.Functions.Worker；使用 Microsoft.Azure.Functions.Worker.Http；使用 Microsoft.DurableTask；使用 Microsoft.DurableTask.Client；使用 Microsoft.Extensions.Logging；公共静态类 DurableFunctionsApp { // HTTP Starter - 触发编排 [函数（“HttpStart”）] 公共静态异步任务 HttpStart( [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "orchestrators/{functionName}")] HttpRequestData req, [DurableClient] DurableTaskClient 客户端，字符串函数名， FunctionContext执行上下文） { 字符串instanceId =等待client.ScheduleNewOrchestrationInstanceAsync(functionName); var logger =executionContext.GetLogger("HttpStart"); logger.LogInformation("已开始编排，ID = '{instanceId}'", instanceId); 返回等待 client.CreateCheckStatusResponseAsync(req, instanceId); } // 协调器函数 [函数（名称（MyOrchestration））] 公共静态异步任务<字符串> MyOrchestration（ [OrchestrationTrigger] TaskOrchestrationContext 上下文） { ILogger 记录器 = context.CreateReplaySafeLogger(nameof(MyOrchestration)); logger.LogInformation("开始编排"); var result1 = wait context.CallActivityAsync(nameof(SayHello), "东京"); var result2 = wait context.CallActivityAsync(nameof(SayHello), "西雅图"); var result3 = wait context.CallActivityAsync(nameof(SayHello), "伦敦"); 返回$“{结果1}，{结果2}，{结果3}”； } // 活动函数 [函数（名称（SayHello））] 公共静态字符串SayHello（[ActivityTrigger]字符串名称，FunctionContextexecutionContext） { var logger =executionContext.GetLogger(nameof(SayHello)); logger.LogInformation("向 {name} 打招呼", name); 返回$“你好{name}！”; } } ### Program.cs 设置使用 Microsoft.Extensions.Hosting； var 主机 = 新的 HostBuilder() .ConfigureFunctionsWorkerDefaults() .Build(); 等待主机.RunAsync(); ## 图案选择指南图案|使用时间 ---|--- **函数链** |顺序步骤，其中每个步骤都依赖于前一个步骤 **扇出/扇入** |并行处理与聚合结果 **异步 HTTP API** |通过 HTTP 状态轮询进行长时间运行的操作 **监控** |具有可配置超时的定期轮询 **人际互动** |工作流程因外部输入/批准而暂停 **聚合器（实体）** |具有操作的状态对象（计数器、帐户）有关详细实现，请参阅 [references/patterns.md]()。 ## 两种方法：基于函数与基于类 ### 基于函数（默认） [函数（名称（MyOrchestration））] 公共静态异步任务<字符串> MyOrchestration（ [OrchestrationTrigger] TaskOrchestrationContext 上下文） { 字符串输入 = context.GetInput()!; 返回等待 context.CallActivityAsync(nameof(MyActivity), input); } [函数（名称（MyActivity））] 公共静态字符串MyActivity（[ActivityTrigger]字符串输入） { return $“已处理：{输入}”； } ### 基于类（带源生成器）需要“Microsoft.DurableTask.Generators”包： [DurableTask(nameof(MyOrchestration))] 公共类 MyOrchestration : TaskOrchestrator { 公共覆盖 async Task RunAsync（TaskOrchestrationContext 上下文，字符串输入） { ILogger 记录器 = context.CreateReplaySafeLogger(); 返回等待 context.CallActivityAsync(nameof(MyActivity), input); } } [DurableTask(名称(MyActivity))] 公共类 MyActivity : TaskActivity { 私有只读 ILogger _logger; // 活动支持 DI - 编排不支持公共 MyActivity(ILogger 记录器) { _logger=记录器； } 公共覆盖 Task RunAsync（TaskActivityContext 上下文，字符串输入） { _logger.LogInformation("处理中：{输入}", input); return Task.FromResult($"已处理：{input}"); } } ## 关键规则 ### 编排决定论编排从历史中重播 - 所有代码都必须是确定性的。当编排恢复时，它会重播所有以前的代码以重建状态。非确定性代码在重播时会产生不同的结果，导致“NonDeterministicOrchestrationException”。 **切勿进行内部编排：** * `DateTime.Now`, `DateTime.UtcNow` → 使用 `context.CurrentUtcDateTime` * `Guid.NewGuid()` → 使用 `context.NewGuid()` * `Random` → 传递来自活动的随机值 * 直接 I/O、HTTP 调用、数据库访问 → 移至活动 * `Thread.Sleep()`, `Task.Delay()` → 使用 `context.CreateTimer()` * 非确定性 LINQ（并行、无序） * `Task.Run()`, `ConfigureAwait(false)` * 静态可变变量 * 可能改变的环境变量 → 作为输入或使用活动传递 **始终安全：** * `context.CallActivityAsync()` * `context.CallSubOrchestrationAsync()` * `context.CallHttpAsync()` * `context.CreateTimer()` * `context.WaitForExternalEvent()` * `context.CurrentUtcDateTime` * `context.NewGuid()` * `context.SetCustomStatus()` * `context.CreateReplaySafeLogger()` ### 非决定论模式（错误与正确） #### 获取当前时间 // 错误 - DateTime.UtcNow 在重播时返回不同的值 [函数（名称（BadOrchestration））] 公共静态异步任务 BadOrchestration（[OrchestrationTrigger] TaskOrchestrationContext 上下文） { 日期时间当前时间 = DateTime.UtcNow; // 不确定性！ if (当前时间.小时 < 12) { 等待 context.CallActivityAsync(nameof(MorningActivity), null); } } // 正确 - context.CurrentUtcDateTime 一致重播 [函数（名称（GoodOrchestration））] 公共静态异步任务GoodOrchestration（[OrchestrationTrigger] TaskOrchestrationContext上下文） { 日期时间 currentTime = context.CurrentUtcDateTime; // 确定性 if (当前时间.小时 < 12) { 等待 context.CallActivityAsync(nameof(MorningActivity), null); } } #### 生成 GUID // 错误 - Guid.NewGuid() 在重播时生成不同的值 [函数（名称（BadOrchestration））] 公共静态异步任务<字符串> BadOrchestration（[OrchestrationTrigger] TaskOrchestrationContext context） { 字符串 orderId = Guid.NewGuid().ToString(); // 不确定性！等待 context.CallActivityAsync(nameof(CreateOrder), orderId); 返回订单ID； } // 正确 - context.NewGuid() 重播相同的值 [函数（名称（GoodOrchestration））] 公共静态异步任务<字符串> GoodOrchestration（[OrchestrationTrigger] TaskOrchestrationContext context） { 字符串 orderId = context.NewGuid().ToString(); // 确定性等待 context.CallActivityAsync(nameof(CreateOrder), orderId); 返回订单ID； } #### 随机数 // 错误 - 随机在重放时产生不同的值 [函数（名称（BadOrchestration））] 公共静态异步任务 BadOrchestration（[OrchestrationTrigger] TaskOrchestrationContext 上下文） { int 延迟 = new Random().Next(1, 10); // 不确定性！等待 context.CreateTimer(context.CurrentUtcDateTime.AddSeconds(delay), CancellationToken.None); } // 正确 - 在活动中生成随机数，传递给协调器 [函数（名称（GetRandomDelay））] public static int GetRandomDelay([ActivityTrigger] 对象？输入) { 返回新的 Random().Next(1, 10); // 活动中正常 } [函数（名称（GoodOrchestration））] 公共静态异步任务GoodOrchestration（[OrchestrationTrigger] TaskOrchestrationContext上下文） { int 延迟 = 等待 context.CallActivityAsync(nameof(GetRandomDelay), null); 等待 context.CreateTimer(context.CurrentUtcDateTime.AddSeconds(delay), CancellationToken.None); } #### 睡觉/延误 // 错误 - Thread.Sleep/Task.Delay 不持久并阻塞 [函数（名称（BadOrchestration））] 公共静态异步任务 BadOrchestration（[OrchestrationTrigger] TaskOrchestrationContext 上下文） { 等待 context.CallActivityAsync(nameof(Step1), null); 等待任务.延迟(60000); // 不耐用！重启时丢失，浪费资源等待 context.CallActivityAsync(nameof(Step2), null); } // 正确 - context.CreateTimer 是持久的 [函数（名称（GoodOrchestration））] 公共静态异步任务GoodOrchestration（[OrchestrationTrigger] TaskOrchestrationContext上下文） { 等待 context.CallActivityAsync(nameof(Step1), null); 等待 context.CreateTimer(context.CurrentUtcDateTime.AddMinutes(1), CancellationToken.None); // 耐用等待 context.CallActivityAsync(nameof(Step2), null); } #### HTTP 调用和 I/O // 错误 - Orchestrator 中的 HttpClient 是不确定的 [函数（名称（BadOrchestration））] 公共静态异步任务<字符串> BadOrchestration（[OrchestrationTrigger] TaskOrchestrationContext context） { 使用 var client = new HttpClient(); var response = wait client.GetStringAsync("https://api.example.com/data"); // 不确定性！返回响应； } // 正确的选项 1 - 使用 CallHttpAsync （内置持久 HTTP） [函数（名称（GoodOrchestration1））] 公共静态异步任务 <字符串> GoodOrchestration1（[OrchestrationTrigger] TaskOrchestrationContext 上下文） { DurableHttpResponse 响应 = 等待 context.CallHttpAsync( HttpMethod.Get, new Uri("https://api.example.com/data")); // 确定性返回响应内容； } // 正确的选项 2 - 将 I/O 移至活动 [函数（名称（FetchData））] 公共静态异步任务 FetchData([ActivityTrigger] string url) { 使用 var client = new HttpClient(); 返回等待 client.GetStringAsync(url); // 活动中正常 } [函数（名称（GoodOrchestration2））] 公共静态异步任务 <字符串> GoodOrchestration2（[OrchestrationTrigger] TaskOrchestrationContext 上下文） { 返回等待 context.CallActivityAsync(nameof(FetchData), "https://api.example.com/data"); } #### 数据库访问 // 错误 - Orchestrator 中的数据库查询 [函数（名称（BadOrchestration））] 公共静态异步任务 BadOrchestration（[OrchestrationTrigger] TaskOrchestrationContext 上下文） { 使用 var conn = new SqlConnection(connectionString); // 不确定性！等待 conn.OpenAsync(); // ... } // 正确 - 活动中的数据库访问 [函数（名称（GetUser））] 公共静态异步任务<用户> GetUser([ActivityTrigger] string userId) { 使用 var conn = new SqlConnection(connectionString); // 活动中正常等待 conn.OpenAsync(); // ... 返回用户； } [函数（名称（GoodOrchestration））] 公共静态异步任务 <用户> GoodOrchestration（[OrchestrationTrigger] TaskOrchestrationContext 上下文） { 字符串 userId = context.GetInput()!; 返回等待 context.CallActivityAsync(nameof(GetUser), userId); } #### 环境变量 // 错误 - 环境变量可能会在重播之间发生变化 [函数（名称（BadOrchestration））] 公共静态异步任务 BadOrchestration（[OrchestrationTrigger] TaskOrchestrationContext 上下文） { string apiEndpoint = Environment.GetEnvironmentVariable("API_ENDPOINT")!; // 可能会改变！等待 context.CallActivityAsync(nameof(CallApi), apiEndpoint); } // 正确 - 将配置作为输入传递 [函数（名称（GoodOrchestration））] 公共静态异步任务GoodOrchestration（[OrchestrationTrigger] TaskOrchestrationContext上下文） { var config = context.GetInput()!; 字符串 apiEndpoint = config.ApiEndpoint; // 根据输入，确定性等待 context.CallActivityAsync(nameof(CallApi), apiEndpoint); } // 也是正确的 - 读取活动中的环境变量 [函数（名称（CallApi））] 公共静态异步任务CallApi（[ActivityTrigger]对象？输入） { string apiEndpoint = Environment.GetEnvironmentVariable("API_ENDPOINT")!; // 活动中正常 // 拨打电话... } #### 集合迭代顺序 // 错误 - 字典迭代顺序可能会有所不同 [函数（名称（BadOrchestration））] 公共静态异步任务 BadOrchestration（[OrchestrationTrigger] TaskOrchestrationContext 上下文） { var items = context.GetInput>()!; foreach (var key in items.Keys) // 不保证顺序！ { 等待 context.CallActivityAsync(nameof(Process), key); } } // 正确 - 使用排序键来确定顺序 [函数（名称（GoodOrchestration））] 公共静态异步任务GoodOrchestration（[OrchestrationTrigger] TaskOrchestrationContext上下文） { var items = context.GetInput>()!; foreach (var key in items.Keys.OrderBy(k => k)) // 保证顺序 { 等待 context.CallActivityAsync(nameof(Process), key); } } ### 登录编排使用 CreateReplaySafeLogger 避免重播期间出现重复的日志条目： [函数（名称（MyOrchestration））] 公共静态异步任务<字符串> MyOrchestration（[OrchestrationTrigger] TaskOrchestrationContext context） { ILogger 记录器 = context.CreateReplaySafeLogger(nameof(MyOrchestration)); logger.LogInformation("编排开始"); // 只记录一次，而不是每次重放 var result = wait context.CallActivityAsync(nameof(MyActivity), "input"); logger.LogInformation("活动已完成，结果：{Result}", result); 返回结果； } ### 错误处理 [函数（名称（OrchestrationWithErrorHandling））] 公共静态异步任务<字符串> OrchestrationWithErrorHandling( [OrchestrationTrigger] TaskOrchestrationContext 上下文） { 字符串输入 = context.GetInput()!; 尝试 { 返回等待 context.CallActivityAsync(nameof(RiskyActivity), input); } 捕获（任务失败异常前） { // 活动失败 - 实施补偿 context.SetCustomStatus(new { Error = ex.Message }); 返回等待 context.CallActivityAsync(nameof(CompensationActivity), input); } } ### 重试策略 var 选项 = 新任务选项 { 重试 = 新的重试策略（最大尝试次数：3，第一重试间隔：TimeSpan.FromSeconds(5), 退避系数：2.0，最大重试间隔：TimeSpan.FromMinutes(1)) }; 等待 context.CallActivityAsync(nameof(UnreliableActivity), 输入, 选项); ## HTTP 管理 API Durable Functions 公开了用于编排管理的内置 HTTP API： ### 创建检查状态响应 [函数（“HttpStart”）] 公共静态异步任务 HttpStart( [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "orchestrators/{functionName}")] HttpRequestData req, [DurableClient] DurableTaskClient 客户端，字符串函数名） { // 解析请求体中的输入字符串？输入=等待新的StreamReader(req.Body).ReadToEndAsync(); 字符串instanceId =等待client.ScheduleNewOrchestrationInstanceAsync(functionName, input); // 返回 202 Accepted with management URLs 作为响应返回等待 client.CreateCheckStatusResponseAsync(req, instanceId); } 回应包括： * `statusQueryGetUri` \- 获取端点以检查状态 * `sendEventPostUri` \- 用于引发事件的 POST 端点 * `terminatePostUri` \- 要终止的 POST 端点 * `purgeHistoryDeleteUri` \- 删除端点以清除历史记录 ### 客户端操作 [DurableClient] DurableTaskClient 客户端 // 安排新的编排字符串instanceId =等待client.ScheduleNewOrchestrationInstanceAsync（“MyOrchestration”，输入）; // 使用自定义实例 ID 进行调度字符串instanceId =等待client.ScheduleNewOrchestrationInstanceAsync( "MyOrchestration", 输入, new StartOrchestrationOptions { InstanceId = "my-custom-id" }); // 获取状态编排元数据？状态 = 等待 client.GetInstanceAsync(instanceId, getInputsAndOutputs: true); // 等待完成编排元数据？结果 = 等待客户端.WaitForInstanceCompletionAsync( instanceId，getInputsAndOutputs：true，cancelToken）； // 引发外部事件等待 client.RaiseEventAsync（instanceId，“ApprovalEvent”，approvalData）； // 终止等待 client.TerminateInstanceAsync(instanceId, "用户已取消"); // 暂停/恢复 wait client.SuspendInstanceAsync(instanceId, "暂停维护"); wait client.ResumeInstanceAsync(instanceId, "恢复操作"); // 清除历史记录等待 client.PurgeInstanceAsync(instanceId); ## 连接和身份验证 ### 连接字符串格式 // 本地模拟器（无身份验证） “端点=http://localhost:8080;身份验证=无” // 具有托管身份的 Azure（推荐用于生产） “端点=https://my-scheduler.region.durabletask.io;身份验证=ManagedIdentity” // 具有特定客户端 ID（用户分配的托管标识）的 Azure “端点=https://my-scheduler.region.durabletask.io;身份验证=ManagedIdentity;ClientId=<客户端id>” 注意：持久任务计划程序仅支持基于身份的身份验证 - 没有带密钥的连接字符串。 ## 使用模拟器进行本地开发 # 启动 Azurite（Azure Functions 必需）蓝铜矿开始 # 拉取并运行 Durable Task Scheduler 模拟器 docker pull mcr.microsoft.com/dts/dts-emulator:最新 docker run -d -p 8080:8080 -p 8082:8082 --name dts-emulator mcr.microsoft.com/dts/dts-emulator:latest # 仪表板位于 http://localhost:8082 # 启动函数应用程序功能开始 ## 持久 HTTP 调用直接从编排进行 HTTP 调用（持久且重放安全）： [函数（名称（CallExternalApi））] 公共静态异步任务<字符串> CallExternalApi（[OrchestrationTrigger] TaskOrchestrationContext context） { // 简单的获取 DurableHttpResponse 响应 =等待 context.CallHttpAsync(HttpMethod.Get, new Uri("https://api.example.com/data")); if (response.StatusCode != HttpStatusCode.OK) { throw new Exception($"API调用失败：{response.StatusCode}"); } 返回响应内容； } // 带有标题和正文 var request = new DurableHttpRequest( HttpMethod.Post, 新的 Uri("https://api.example.com/data")) { headers = { ["Content-Type"] = "application/json" }, 内容= JsonSerializer.Serialize（有效负载） }; DurableHttpResponse 响应 = 等待 context.CallHttpAsync(request); // 具有托管身份认证 var request = new DurableHttpRequest( HttpMethod.Get, 新 Uri("https://management.azure.com/...")) { TokenSource = new ManagedIdentityTokenSource("https://management.azure.com/.default") }; ＃＃参考 * **[patterns.md]()** \- 详细的模式实现（扇出/扇入、人机交互、实体、子编排、监视器） * **[setup.md]()** \- Azure 持久任务计划程序预配、部署和项目模板

耐用功能 Dotnet

包含技能