Hubs de tarefas

Um hub de tarefas é uma representação do estado atual do aplicativo no armazenamento, incluindo todo o trabalho pendente. Enquanto um aplicativo é executado, o hub de tarefas armazena continuamente o progresso das funções de orquestração, atividade e entidade. Essa abordagem garante que o aplicativo possa retomar o processamento do ponto em que parou, caso seja reiniciado após uma interrupção temporária. Um hub de tarefas também permite que os aplicativos dimensionem os recursos de computação dinamicamente.

Este artigo explica o que um hub de tarefas armazena, como configurar e nomear hubs de tarefas, como criá-los e gerenciá-los com back-ends de armazenamento diferentes e como o processo dos hubs de tarefas funciona internamente.

Conceitualmente, um hub de tarefas armazena as seguintes informações:

• Os estados de instância de todas as instâncias de orquestração e entidade.
• As mensagens a serem processadas, incluindo:
  • Todas as mensagens de atividade que representam atividades aguardando execução.
  • Quaisquer mensagens de instância que estejam aguardando para serem entregues às instâncias.

As mensagens de atividade são sem estado e podem ser processadas em qualquer lugar. As mensagens de instância precisam ser entregues a uma instância com estado específico (orquestração ou entidade), identificada por sua ID de instância.

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "hubName": "MyTaskHub"
    }
  }
}

{
  "durableTask": {
    "hubName": "MyTaskHub"
  }
}

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "hubName": "%MyTaskHub%"
    }
  }
}

{
  "durableTask": {
    "hubName": "%MyTaskHub%"
  }
}

{
  "IsEncrypted": false,
  "Values": {
    "MyTaskHub" : "samplehubname"
  }
}

[FunctionName("HttpStart")]
public static async Task<HttpResponseMessage> Run(
    [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}")] HttpRequestMessage req,
    [DurableClient(TaskHub = "%MyTaskHub%")] IDurableOrchestrationClient starter,
    string functionName,
    ILogger log)
{
    // Function input comes from the request content.
    object eventData = await req.Content.ReadAsAsync<object>();
    string instanceId = await starter.StartNewAsync(functionName, eventData);

    log.LogInformation($"Started orchestration with ID = '{instanceId}'.");

    return starter.CreateCheckStatusResponse(req, instanceId);
}

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

@FunctionName("HttpStart")
public HttpResponseMessage httpStart(
        @HttpTrigger(name = "req", route = "orchestrators/{functionName}") HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext", taskHub = "%MyTaskHub%") DurableClientContext durableContext,
        @BindingName("functionName") String functionName,
        final ExecutionContext context) {

    DurableTaskClient client = durableContext.getClient();
    String instanceId = client.scheduleNewOrchestrationInstance(functionName);
    context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);
    return durableContext.createCheckStatusResponse(req, instanceId);
}

Usar vários aplicativos com hubs de tarefas separados

Cada aplicativo que compartilha um back-end deve se conectar ao seu próprio hub de tarefas para evitar conflitos. Se vários aplicativos usam o mesmo hub de tarefas, eles competem por mensagens, o que pode resultar em um comportamento indefinido, incluindo orquestrações ficando inesperadamente paralisadas. Um único back-end pode conter vários hubs de tarefas; configurar cada aplicativo com o seu próprio.

Gerenciamento do hub de tarefas do Agendador de Tarefas Durável

Esta seção aborda como criar e gerenciar hubs de tarefas ao usar o back-end do Agendador de Tarefas Duráveis. Crie os recursos do agendador e do hub de tarefas explicitamente antes que seu aplicativo os use.

Criar um agendador e um hub de tarefas

Crie um agendador e um hub de tarefas usando o portal Azure, Azure CLI, Azure Resource Manager (ARM) ou Bicep.

@description('The name of the Durable Task Scheduler.')
param schedulerName string

@description('The name of the task hub.')
param taskHubName string

@description('The location for all resources.')
param location string = resourceGroup().location

resource scheduler 'Microsoft.DurableTask/schedulers@2025-04-01-preview' = {
  name: schedulerName
  location: location
  properties: {
    ipAllowlist: [
      '0.0.0.0/0'
    ]
    sku: {
      name: 'Dedicated'
      capacity: 1
    }
  }
}

resource taskHub 'Microsoft.DurableTask/schedulers/taskHubs@2025-04-01-preview' = {
  parent: scheduler
  name: taskHubName
  properties: {}
}

Os exemplos anteriores usam o SKU Dedicado. O Agendador de Tarefas Duráveis também oferece um SKU de Consumo. Para obter mais informações sobre como gerenciar recursos do Agendador de Tarefas Duráveis, consulte Desenvolver com o Agendador de Tarefas Duráveis.

Configurar a autenticação baseada em identidade

O Agendador de Tarefas Duráveis dá suporte apenas à autenticação de identidade gerenciada. Ele não dá suporte a cadeias de conexão com chaves de armazenamento. Atribua a função RBAC (controle de acesso baseado em função) apropriada a uma identidade gerenciada e configure seu aplicativo para usar essa identidade.

As seguintes funções estão disponíveis:

Use identidades gerenciadas atribuídas pelo usuário quando possível porque elas não estão vinculadas ao ciclo de vida do aplicativo e você pode reutilizá-las depois que o aplicativo for removido.

PUT https://management.azure.com/subscriptions/SUBSCRIPTION_ID/resourceGroups/RESOURCE_GROUP/providers/Microsoft.DurableTask/schedulers/SCHEDULER_NAME/providers/Microsoft.Authorization/roleAssignments/ROLE_ASSIGNMENT_ID?api-version=2022-04-01

{
  "properties": {
    "roleDefinitionId": "/subscriptions/SUBSCRIPTION_ID/providers/Microsoft.Authorization/roleDefinitions/ROLE_DEFINITION_ID",
    "principalId": "MANAGED_IDENTITY_PRINCIPAL_ID",
    "principalType": "ServicePrincipal"
  }
}

@description('The principal ID of the managed identity.')
param identityPrincipalId string

@description('The Durable Task Data Contributor role definition ID.')
var durableTaskDataContributorRoleId = subscriptionResourceId(
  'Microsoft.Authorization/roleDefinitions',
  'ROLE_DEFINITION_ID'
)

resource roleAssignment 'Microsoft.Authorization/roleAssignments@2022-04-01' = {
  name: guid(scheduler.id, identityPrincipalId, durableTaskDataContributorRoleId)
  scope: scheduler
  properties: {
    roleDefinitionId: durableTaskDataContributorRoleId
    principalId: identityPrincipalId
    principalType: 'ServicePrincipal'
  }
}

Depois de atribuir a identidade, adicione as seguintes variáveis de ambiente ao seu aplicativo:

Para obter detalhes completos da configuração de identidade, consulte Configurar a identidade gerenciada para o Agendador de Tarefas Duráveis.

Itens de trabalho

As mensagens de atividade e as mensagens de instância no hub de tarefas representam o trabalho que o aplicativo precisa processar. Enquanto o aplicativo está em execução, ele busca continuamente itens de trabalho do hub de tarefas. Cada item de trabalho processa uma ou mais mensagens. Há dois tipos de itens de trabalho:

• Itens de trabalho de atividade: executar uma função de atividade para processar uma mensagem de atividade.
• Itens de trabalho do orquestrador: executam uma função de orquestrador ou entidade para processar uma ou mais mensagens de instância.

Os trabalhadores podem processar vários itens de trabalho ao mesmo tempo, sujeitos aos limites de simultaneidade por trabalho configurados.

Depois que um trabalho conclui um item de trabalho, ele confirma os efeitos de volta para o hub de tarefas. Esses efeitos variam de acordo com o tipo de função que foi executada:

• Uma função de atividade concluída cria uma mensagem de instância contendo o resultado, endereçada à instância do orquestrador pai.
• Uma função de orquestrador concluída atualiza o estado e o histórico da orquestração e pode criar novas mensagens.
• Uma função de entidade concluída atualiza o estado da entidade e também pode criar novas mensagens de instância.

Para orquestrações, cada item de trabalho representa um episódio da execução dessa orquestração. Um episódio é uma rodada do orquestrador em execução, processando resultados disponíveis e, em seguida, pausando até que o próximo resultado chegue. Por exemplo, um episódio começa quando a orquestração começa, quando uma atividade é concluída e retorna um resultado ou quando um evento externo chega. O episódio termina quando o orquestrador termina ou chega a um ponto em que deve aguardar novas mensagens.

Exemplo de execução

Considere uma orquestração do tipo fan-out-fan-in que inicia duas atividades em paralelo e aguarda que ambas sejam concluídas:

[FunctionName("Example")]
public static async Task Run([OrchestrationTrigger] IDurableOrchestrationContext context)
{
    Task t1 = context.CallActivityAsync<int>("MyActivity", 1);
    Task t2 = context.CallActivityAsync<int>("MyActivity", 2);
    await Task.WhenAll(t1, t2);
}

module.exports = df.orchestrator(function*(context){
    const tasks = [];
    tasks.push(context.df.callActivity("MyActivity", 1));
    tasks.push(context.df.callActivity("MyActivity", 2));
    yield context.df.Task.all(tasks);
});

def orchestrator_function(context: df.DurableOrchestrationContext):
    tasks = []
    tasks.append(context.call_activity("MyActivity", 1))
    tasks.append(context.call_activity("MyActivity", 2))
    yield context.task_all(tasks)

param($Context)

$Tasks = @()

$Tasks += Invoke-DurableActivity -FunctionName 'MyActivity' -Input 1 -NoWait
$Tasks += Invoke-DurableActivity -FunctionName 'MyActivity' -Input 2 -NoWait

Wait-DurableTask -Task $Tasks

@FunctionName("Example")
public void exampleOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    Task<Void> t1 = ctx.callActivity("MyActivity", 1);
    Task<Void> t2 = ctx.callActivity("MyActivity", 2);
    ctx.allOf(List.of(t1, t2)).await();
}

using Microsoft.DurableTask;

public class Example : TaskOrchestrator<object?, object?>
{
    public override async Task<object?> RunAsync(TaskOrchestrationContext context, object? input)
    {
        Task t1 = context.CallActivityAsync("MyActivity", 1);
        Task t2 = context.CallActivityAsync("MyActivity", 2);
        await Task.WhenAll(t1, t2);
        return null;
    }
}

from durabletask import task

def example_orchestrator(ctx: task.OrchestrationContext, _):
    t1 = ctx.call_activity("my_activity", input=1)
    t2 = ctx.call_activity("my_activity", input=2)
    yield task.when_all([t1, t2])

import com.microsoft.durabletask.TaskOrchestration;
import com.microsoft.durabletask.TaskOrchestrationContext;

public class Example implements TaskOrchestration {
    @Override
    public void run(TaskOrchestrationContext ctx) {
        Task<Void> t1 = ctx.callActivity("MyActivity", 1, Void.class);
        Task<Void> t2 = ctx.callActivity("MyActivity", 2, Void.class);
        ctx.allOf(List.of(t1, t2)).await();
    }
}

Depois que essa orquestração é iniciada por um cliente, o aplicativo o processa como uma sequência de itens de trabalho. Cada item de trabalho concluído atualiza o estado do hub de tarefas quando ele é confirmado. Estas são as etapas:

1. Um cliente solicita iniciar uma nova orquestração com a ID da instância "123". Depois que o cliente concluir essa solicitação, o hub de tarefas conterá um espaço reservado para o estado de orquestração e uma mensagem de instância:

   

   O rótulo ExecutionStarted é um dos muitos tipos de eventos de histórico que identificam os vários tipos de mensagens e eventos que participam do histórico de uma orquestração.

2. Um trabalhador executa um item de trabalho do orquestrador para processar a ExecutionStarted mensagem. Ele chama a função orquestradora que inicia a execução do código de orquestração. Esse código agenda duas atividades e, em seguida, interrompe a execução quando aguarda os resultados.

   

   O estado de runtime é agora Running, e o histórico registra este primeiro episódio: o orquestrador começou, a execução começou, duas tarefas foram agendadas, e o orquestrador concluiu o episódio.

3. Um trabalhador executa um item de trabalho de atividade para processar uma das TaskScheduled mensagens. Ele chama a função de atividade com a entrada "2". Quando a função de atividade é concluída, ela cria uma TaskCompleted mensagem que contém o resultado.

   

4. Um trabalhador executa um item de trabalho do orquestrador para processar a TaskCompleted mensagem. Se a orquestração ainda estiver armazenada em cache na memória, ela poderá apenas retomar a execução. Caso contrário, o trabalho primeiro repete a história para recuperar o estado atual da orquestração. Em seguida, continua a orquestração, entregando o resultado da atividade. Depois de receber esse resultado, a orquestração ainda aguarda o resultado da outra atividade, portanto, ela para mais uma vez a execução.

   

   O histórico registra o segundo episódio: a tarefa concluída e o orquestrador pausado novamente.

5. Um trabalho executa um item de trabalho de atividade para processar a mensagem TaskScheduled restante. Ele chama a função de atividade com a entrada "1".

   

6. Um trabalhador executa outro item de trabalho do orquestrador para processar a TaskCompleted mensagem. Depois de receber este segundo resultado, a orquestração é concluída.

   

   O estado de runtime é agora Completed, e o histórico registra o terceiro e último episódio: a segunda tarefa concluída e a execução concluída.

Representação no sistema de armazenamento

Provedor de Agendador de Tarefas Duráveis

O Agendador de Tarefas Duráveis é um provedor de back-end totalmente gerenciado que armazena todo o estado do hub de tarefas internamente. Ao contrário dos provedores de armazenamento BYO (bring-your-own), você não precisa configurar nem gerenciar nenhuma infraestrutura de armazenamento subjacente. Cada recurso do agendador (Microsoft.DurableTask/schedulers) tem recursos de computação e memória dedicados e pode conter um ou mais hubs de tarefas (Microsoft.DurableTask/schedulers/taskHubs).

Como o Agendador de Tarefas Duráveis gerencia o armazenamento internamente, você não pode inspecionar diretamente os dados subjacentes. Em vez disso, use o painel Agendador de Tarefas Duráveis para monitorar e consultar instâncias de orquestração.