Durable Functions (Azure Functions) 的绑定

Durable Functions 支持 Azure Functions 的两个版本的 Python 编程模型。 由于 Python v2 是推荐的版本，因此本文中的示例仅支持此版本。

先决条件

• Durable Functions SDK，即 Python 包索引 (PyPI) 中的包 azure-functions-durable，版本 1.2.2 或更高版本
• 扩展包 版本 4.x（或更高版本），已在 host.json 项目文件中设置

可以在 Durable Functions SDK for Python 存储库中提供反馈和建议。

在 .NET 中开发函数时，可以使用 OrchestrationTriggerAttribute .NET 属性配置业务流程触发器。

对于 Java，可以使用 @DurableOrchestrationTrigger 批注来配置业务流程触发器。

使用 Node.js 编程模型版本 4 开发函数时，请从app模块导入@azure/functions npm对象。 然后，直接在函数代码中调用 app.orchestration Durable Functions API 的方法。 此方法将协调器函数注册到 Durable Functions 框架中。

编写业务流程协调程序函数时，可以在 bindings 文件的数组中使用以下 JSON 对象定义业务流程触发器：

{
    "name": "<name-of-input-parameter-in-function-signature>",
    "orchestration": "<optional-name-of-orchestration>",
    "type": "orchestrationTrigger",
    "direction": "in"
}

该值 orchestration 是客户端想要启动协调函数新实例时必须使用的协调程序名称。 此属性是可选的。 如果未指定，则使用函数的名称。

使用 Python v2 编程模型时，可以直接在 Python 函数代码中使用 orchestration_trigger 修饰器来定义业务流程触发器。

在 v2 模型中，可以从 实例 DFApp访问 Durable Functions 触发器和绑定。 可以使用此子类 FunctionApp 导出特定于 Durable Functions 的修饰器。

用于定义触发器的属性取决于是在 与 Functions 主机进程相同的进程中 运行 C# 函数，还是在 独立工作进程中运行 C# 函数。

[FunctionName("HelloWorld")]
public static string RunOrchestrator([OrchestrationTrigger] IDurableOrchestrationContext context)
{
    string name = context.GetInput<string>();
    return $"Hello {name}!";
}

[Function("HelloWorld")]
public static string RunOrchestrator([OrchestrationTrigger] TaskOrchestrationContext context, string name)
{
    return $"Hello {name}!";
}

const { app } = require('@azure/functions');
const df = require('durable-functions');

df.app.orchestration('helloOrchestrator', function* (context) {
    const name = context.df.getInput();
    return `Hello ${name}`;
});

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

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

@myApp.orchestration_trigger(context_name="context")
def my_orchestrator(context):
    result = yield context.call_activity("Hello", "Tokyo")
    return result

param($Context)

$InputData = $Context.Input
$InputData

@FunctionName("HelloWorldOrchestration")
public String helloWorldOrchestration(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    return String.format("Hello %s!", ctx.getInput(String.class));
}

大多数编排器函数调用活动函数。 以下代码提供了一个 Hello World 示例，演示如何调用活动函数：

const { app } = require('@azure/functions');
const df = require('durable-functions');

const activityName = 'hello';

df.app.orchestration('helloOrchestrator', function* (context) {
    const name = context.df.getInput();
    const result = yield context.df.callActivity(activityName, name);
    return result;
});

@FunctionName("HelloWorld")
public String helloWorldOrchestration(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    String input = ctx.getInput(String.class);
    String result = ctx.callActivity("SayHello", input, String.class).await();
    return result;
}

使用 ActivityTriggerAttribute .NET 属性配置活动触发器。

使用 @DurableActivityTrigger 批注配置活动触发器。

若要注册活动函数，请从app模块导入@azure/functions npm对象。 然后，直接在函数代码中调用 app.activity Durable Functions API 的方法。

若要定义活动触发器，请在 bindings数组中使用以下 JSON 对象：

{
    "name": "<name-of-input-parameter-in-function-signature>",
    "activity": "<optional-name-of-activity>",
    "type": "activityTrigger",
    "direction": "in"
}

该值 activity 是活动的名称。 此值是协调器函数用于调用此活动函数的名称。 此属性是可选的。 如果未指定，则使用函数的名称。

可以直接在 Python 函数代码中使用 activity_trigger 修饰器来定义活动触发器。

const { app } = require('@azure/functions');
const df = require('durable-functions');
const activityName = 'hello';
df.app.activity(activityName, {
    handler: (input) => {
        return `Hello, ${input}`;
    },
});

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

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

@myApp.activity_trigger(input_name="myInput")
def my_activity(myInput: str):
    return "Hello " + myInput

param($name)

"Hello $name!"

@FunctionName("SayHello")
public String sayHello(@DurableActivityTrigger(name = "name") String name) {
    return String.format("Hello %s!", name);
}

例如，活动函数可以从协调器函数接收输入。 然后，活动函数可以将该输入作为消息发送到 Azure 事件中心。

const { app } = require('@azure/functions');
const df = require('durable-functions');

df.app.orchestration('helloOrchestrator', function* (context) {
    const input = context.df.getInput();
    yield context.df.callActivity('sendToEventHub', input);
    return `Message sent: ${input}`;
});

const { EventHubProducerClient } = require("@azure/event-hubs");
const connectionString = process.env.EVENT_HUB_CONNECTION_STRING;
const eventHubName = process.env.EVENT_HUB_NAME;

df.app.activity("sendToEventHub", {
    handler: async (message, context) => {
        const producer = new EventHubProducerClient(connectionString, eventHubName);
        try {
            const batch = await producer.createBatch();
            batch.tryAdd({ body: message });
            await producer.sendBatch(batch);
            context.log(`Message sent to Event Hubs: ${message}`);
        } catch (err) {
            context.log.error("Failed to send message to Event Hubs:", err);
            throw err;
        } finally {
            await producer.close();
        }
    },
});

app.storageQueue('helloQueueStart', {
    queueName: 'start-orchestration',
    extraInputs: [df.input.durableClient()],
    handler: async (message, context) => {
        const client = df.getClient(context);
        const orchestratorName = message.orchestratorName || 'helloOrchestrator';
        const input = message.input || null;
        const instanceId = await client.startNew(orchestratorName, { input });
        context.log(`Started orchestration with ID = '${instanceId}'`);
    },
});

可以使用 DurableClientAttribute 属性（在 Durable Functions 1.x 中为 OrchestrationClientAttribute）绑定到编排客户端。

可以使用@DurableClientInput注解绑定业务流程客户端。

若要注册客户端函数，请从app模块导入@azure/functions npm对象。 然后，调用特定于触发器类型的 Durable Functions API 方法。 例如，对于 HTTP 触发器，调用 app.http 该方法。 对于队列触发器，调用该方法app.storageQueue。

若要定义持久客户端触发器，请在 bindings数组中使用以下 JSON 对象：

{
    "name": "<name-of-input-parameter-in-function-signature>",
    "taskHub": "<optional-name-of-task-hub>",
    "connectionName": "<optional-name-of-connection-string-app-setting>",
    "type": "orchestrationClient",
    "direction": "in"
}

• 当多个函数应用共享同一存储帐户，但需要彼此隔离时，将使用 taskHub 此属性。 如果未指定此属性，则使用 host.json 中的默认值。 此值必须与目标业务流程协调程序函数使用的值匹配。
• 该值 connectionName 是包含存储帐户连接字符串的应用设置的名称。 此连接字符串所表示的存储帐户必须与目标业务流程协调程序函数使用的存储帐户相同。 如果未指定此属性，将使用函数应用的默认存储帐户连接字符串。

可以直接在 Python 函数代码中使用 durable_client_input 修饰器来定义持久客户端触发器。

通常情况下，你会绑定到 IDurableClient 实现（在 Durable Functions 1.x 中为 DurableOrchestrationClient），从而可以完全访问 Durable Functions 支持的所有编排客户端 API。

通常绑定到 DurableClientContext 类。

必须使用特定于语言的 SDK 来访问客户端对象。

const { app } = require('@azure/functions');
const df = require('durable-functions');

app.storageQueue('helloQueueStart', {
    queueName: 'start-orchestration',
    extraInputs: [df.input.durableClient()],
    handler: async (message, context) => {
        const client = df.getClient(context);
        const orchestratorName = message.orchestratorName || 'helloOrchestrator';
        const input = message.input || null;
        const instanceId = await client.startNew(orchestratorName, { input });
        context.log(`Started orchestration with ID = '${instanceId}' from queue message.`);
    },
});

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

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

@myApp.queue_trigger(
    arg_name="msg",
    queue_name="start-orchestration",
    connection="AzureWebJobsStorage"
)
@myApp.durable_client_input(client_name="client")
async def client_function(msg: func.QueueMessage, client: df.DurableOrchestrationClient):
    input_data = msg.get_body().decode("utf-8")
    await client.start_new("my_orchestrator", None, input_data)
    return None

function.json

{
  "bindings": [
    {
      "name": "InputData",
      "type": "queueTrigger",
      "queueName": "durable-function-trigger",
      "direction": "in"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ]
}

run.ps1

param([string]$InputData, $TriggerMetadata)

$InstanceId = Start-DurableOrchestration -FunctionName 'HelloWorld' -Input $InputData

@FunctionName("QueueStart")
public void queueStart(
        @QueueTrigger(name = "input", queueName = "durable-function-trigger", connection = "Storage") String input,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    // Orchestration input comes from the queue message content.
    durableContext.getClient().scheduleNewOrchestrationInstance("HelloWorld", input);
}

使用 EntityTriggerAttribute .NET 属性配置实体触发器。

若要注册实体触发器，请从app模块导入@azure/functions npm对象。 然后，直接在函数代码中调用 app.entity Durable Functions API 的方法。

const df = require('durable-functions');
df.app.entity('counter', (context) => {
    const currentValue = context.df.getState(() => 0);
    switch (context.df.operationName) {
        case 'add':
            context.df.setState(currentValue + context.df.getInput());
            break;
        case 'reset':
            context.df.setState(0);
            break;
        case 'get':
            context.df.return(currentValue);
            break;
    }
});

可以直接在 Python 函数代码中使用 entity_trigger 修饰器来定义实体触发器。

可以在 .NET 类库函数中使用 DurableClientAttribute .NET 属性绑定到实体客户端。

与其注册实体客户端，不如使用 signalEntity 或 callEntity 从任何已注册的函数调用实体触发器方法。

• 在队列触发的函数中，可以使用client.signalEntity。

  const { app } = require('@azure/functions');
  const df = require('durable-functions');
  app.storageQueue('helloQueueStart', {
      queueName: 'start-orchestration',
      extraInputs: [df.input.durableClient()],
      handler: async (message, context) => {
          const client = df.getClient(context);
          const entityId = new df.EntityId('counter', 'myCounter');
          await client.signalEntity(entityId, 'add', 5);
      },
  });
  

• 在编排程序函数中，你可以使用context.df.callEntity:

  const { app } = require('@azure/functions');
  const df = require('durable-functions');
  df.app.orchestration('entityCaller', function* (context) {
      const entityId = new df.EntityId('counter', 'myCounter');
      yield context.df.callEntity(entityId, 'add', 5);
      yield context.df.callEntity(entityId, 'add', 5);
      const result = yield context.df.callEntity(entityId, 'get');
      return result;
  });
  

可以直接在 Python 函数代码中使用 durable_client_input 修饰器来定义实体客户端。