Filtrowanie elementów pracy w Trwałym Harmonogramie Zadań

Filtrowanie zadań roboczych to funkcja routingu w Trwałym Planiście Zadań, która dostarcza każdy element pracy tylko pracownikom, którzy zadeklarują, że hostują dopasowaną orkiestrację lub aktywność. Procesy produkcyjne często łączą etapy z różnymi wymaganiami obliczeniowymi, takimi jak lekka koordynacja, działania na dużą skalę oraz specjalistyczne przetwarzanie. Bez możliwości samodzielnego kierowania każdego kroku, te kroki konkurują o tę samą przepustowość lub zmuszają do budowy osobnej infrastruktury koordynacyjnej dla każdego z nich.

To trasowanie opiera się na topologii płaskiej sieci: każdy węzeł roboczy nawiązuje połączenie wychodzące z harmonogramem zamiast bezpośrednio z innymi węzłami roboczymi. W rezultacie planista koordynuje trasowanie za ciebie, bez wykrywania usług, portów przychodzących czy dedykowanej warstwy komunikacyjnej do samodzielnego budowania i obsługi.

W tym artykule dowiesz się, kiedy używać filtrowania elementów roboczych i jak harmonogram zadań kieruje każdy krok do zasobu obliczeniowego, który najlepiej do niego pasuje.

Kiedy stosować filtrowanie zadań roboczych

Filtrowanie zadań jest najbardziej przydatne, gdy etapy jednego workflow mają różne wymagania w czasie działania. Bez niego każdy krok działa na tej samej puli obliczeniowej, więc dostosowujesz cały workflow do najbardziej wymagającego etapu, a specjalistyczne lub zawsze aktywne komponenty konkurują ze wszystkim innym o pojemność. Poniższe scenariusze pokazują, jak kierowanie każdego etapu do odpowiednio dobranej mocy obliczeniowej rozwiązuje ten problem.

Łącz zasoby obliczeniowe stale dostępne z zasobami skalowanymi do zera

Niektóre części aplikacji działają nieprzerwanie, podczas gdy inne nie powinny kosztować nic podczas bezczynności. Dzięki filtrowaniu zadań możesz umieścić każdy etap na poziomie obliczeniowym, który odpowiada jego obciążeniu.

• Uruchamiaj komponenty działające stale na gotowych zasobach obliczeniowych, takich jak Azure Kubernetes Service (AKS), aby orkiestracje były uruchamiane natychmiast.
• Uruchamiaj komponenty o nierównomiernym obciążeniu, sterowane zdarzeniami, w środowisku bezserwerowym, takim jak Azure Container Apps lub Azure Functions, które skalują się od zera, gdy pojawiają się zadania, i z powrotem do zera, gdy obciążenie ustaje.

Filtrowanie gwarantuje, że każda warstwa pobiera tylko zadania, które obsługuje, dzięki czemu komponenty o skokowym obciążeniu nigdy nie wypierają tych stale działających, a zasoby obliczeniowe skalowane do zera są uruchamiane tylko dla działań, za które odpowiadają. Szczegóły dotyczące autoskalowania zobacz Configure autoscale for Azure Container Apps.

Stale uruchomiony proces roboczy orkiestratora w usłudze Azure Kubernetes Service oraz skalowany do zera proces roboczy działań w usłudze Azure Container Apps lub Azure Functions nawiązują połączenie wychodzące z Durable Task Scheduler.

Obliczenia odpowiedniej wielkości na każdy krok aktywności

Gdy jeden etap zużywa znacznie więcej pamięci lub procesora niż pozostały, dostosowanie całego wdrożenia pod ten etap marnuje zasoby. Filtrowanie zadań pozwala alokować zasoby na każdy etap w ramach jednego workflow.

• Przypisz wymagającemu działaniu — na przykład etapowi agregacji danych, który ładuje duży zbiór danych do pamięci — oddzielny proces roboczy w większym rozmiarze SKU i zarejestruj tam tylko to działanie.
• Rób co drugi krok na mniejszych, tańszych pracownikach.

Każdego pracownika dostosowujesz do wykonywanej pracy, zamiast skalować jedno jednolite wdrożenie, aby spełnić wymagania największego etapu.

Mali pracownicy SKU obsługują lekkie zadania, a duży pracownik SKU wykonuje zadania wymagające dużej ilości pamięci, a wszyscy pracownicy są kierowani przez Trwały Harmonogram Zadań.

Kieruj obciążenia robocze do wyspecjalizowanego sprzętu

Pojedynczy etap w przepływie pracy może wymagać specjalistycznego sprzętu, którego reszta aplikacji nie wymaga. Na przykład workflow działa głównie w Azure Functions, ale jeden etap wymaga GPU do uruchomienia modelu AI. Planista wysyła ten krok do pracownika działającego w Azure Container Apps na infrastrukturze obsługującej GPU, a proces pracy jest kontynuowany w Azure Functions.

Wdroż pracownika wspieranego przez GPU, który rejestruje tylko aktywność GPU, a resztę pracy zachowaj na standardowym poziomie obliczeniowym. Planista kieruje zadania każdej aktywności do pracownika stworzonego dla niej, więc infrastruktura GPU wykonuje tylko tę pracę, która jej potrzebuje.

Klient uruchamia workflow w Trwałym Harmonogramie Zadań. Azure Functions obsługuje orkiestrację i działania CPU, a pracownik Azure Container Apps z GPU obsługuje aktywność GPU.

Jak działa filtrowanie zadań

Trwały harmonogram zadań dzieli wykonanie orkiestracji na odrębne jednostki pracy zwane elementami pracy. Istnieją trzy główne typy zadań:

• Zadania orkiestracji: Uruchamiaj kod orchestratora, który koordynuje przepływ pracy.
• Zadania związane z czynnością: Uruchamiaj poszczególne funkcje czynności, które wykonują faktyczną pracę, takie jak walidacja zamówienia lub jego wysyłka.
• Elementy pracy podmiotów: Wykonuj operacje na trwałych jednostkach, które harmonogram kieruje według nazwy zarejestrowanej jednostki.

Proces roboczy to każdy proces, który łączy się z planistą zadań, aby pobierać i wykonywać elementy pracy. Pracownik może gościć orkiestracje, działania, podmioty oraz dowolną ich kombinację.

Dostarczenie dowolnego zadania do dowolnego powiązanego pracownika działa dobrze, gdy pojedynczy pracownik lub grupa identycznych pracowników rejestruje każde zadanie w procesie. Filtrowanie zadań daje alternatywę: możesz kierować konkretne zadania do konkretnych pracowników. Każdy pracownik rejestruje tylko zadania, które hostuje, a harmonogram kieruje do niego tylko dopasowane zadania.

Gdy pracownik zaczyna, rejestruje swoje lokalne zadania i wybiera filtrowanie. SDK generuje listę filtrów z zarejestrowanych zadań i wysyła ją do planisty, gdy pracownik się łączy. Mechanizm planowania tworzy oddzielne kolejki dla każdego filtra i kieruje każdy element roboczy do kolejki, która odpowiada typowi zadania. Pracownik pobiera tylko z kolejek odpowiadających jego filtrom.

Na przykład w procesie przetwarzania zamówień:

1. Pracownik Orchestratora rejestruje i otrzymuje OrderProcessingOrchestration tylko zadania orkiestracyjne.
2. Pracownik Walidatora rejestruje i otrzymuje ValidateOrder tylko ValidateOrder zadania związane z aktywnością.
3. Pracownik Nadawcy rejestruje ShipOrder się i otrzymuje wyłącznie ShipOrder zadania związane z aktywnością.

Trasowanie jest ścisłe. Nie ma mechanizmu przełączenia na innego workera:

• Jeśli żaden worker nie zarejestruje danego zadania, jego elementy pracy pozostaną Pending, aż połączy się pasujący worker.
• Orkiestracja, która zależy od tego zadania, pozostaje w stanie Running do czasu ukończenia oczekującego elementu roboczego.

Ta rygorystyka zapewnia, że planista nigdy nie przekazuje zadania pracownikowi, który nie jest w stanie go przetworzyć.

Obsługa filtrowania elementów pracy w Durable Task SDK

Trwały zestaw SDK zadań	Filtrowanie elementów roboczych
.NET	Available
Java	Available
Python	Available
Go (beta)	Available
JavaScript	Jeszcze niedostępne
PowerShell	Jeszcze niedostępne

Włącz filtrowanie zadań

Każdy pracownik rejestruje tylko zadania, które hostuje, a SDK wykorzystuje tę rejestrację do generowania filtrów zadań pracownika. Włączanie filtrowania zadań różni się nieco w zależności od SDK.

.NET

System automatycznie generuje filtry zadań z rejestrów zadań każdego pracownika, więc nie musisz wykonywać wyraźnych wywołań.

builder.Services.AddDurableTaskWorker()
    .AddTasks(registry =>
    {
        registry.AddAllGeneratedTasks();
    })
    .UseDurableTaskScheduler(connectionString);

Uwaga / Notatka

W SDK .NET Durable Task rejestr automatycznie generuje filtry zadań. Aby zastąpić wygenerowane filtry lub wyłączyć filtrowanie, wywołaj odpowiednio UseWorkItemFilters(customFilters) lub UseWorkItemFilters(null).

Java

Wezwij useWorkItemFilters() pracownika budowlanego. SDK generuje filtry z zarejestrowanych zadań.

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions
        .createWorkerBuilder(connectionString)
        .addActivity(new TaskActivityFactory() {
            // ...
        })
        .useWorkItemFilters() // auto-generate from registered tasks
        .build();

Aby podać jawne filtry zamiast generować je automatycznie z rejestru, należy przekazać WorkItemFilter:

WorkItemFilter filter = WorkItemFilter.newBuilder()
        .addOrchestration("OrderProcessingOrchestration")
        .addActivity("ValidateOrder")
        .build();

builder.useWorkItemFilters(filter);

Python

Zadzwoń use_work_item_filters() do pracownika po zarejestrowaniu jego zadań. SDK generuje filtry z zarejestrowanych zadań.

worker.add_orchestrator(order_processing_orchestrator)
worker.add_activity(validate_order)
worker.use_work_item_filters()  # auto-generate from registered tasks

Aby dostarczyć filtry jawne zamiast automatycznie generować je z rejestru, należy przejść WorkItemFilters:

from durabletask.worker import WorkItemFilters, OrchestrationWorkItemFilter, ActivityWorkItemFilter

worker.use_work_item_filters(WorkItemFilters(
    orchestrations=[
        OrchestrationWorkItemFilter(name="order_processing_orchestrator"),
    ],
    activities=[
        ActivityWorkItemFilter(name="validate_order"),
    ],
))

Go

Przekaż client.WithAutoWorkItemFilters() podczas tworzenia workera. SDK tworzy filtry na podstawie dostarczonego rejestru:

package main

import (
	"github.com/microsoft/durabletask-go/api"
	"github.com/microsoft/durabletask-go/client"
	"github.com/microsoft/durabletask-go/durabletaskscheduler"
	"github.com/microsoft/durabletask-go/task"
)

func newFilteredWorker(options *durabletaskscheduler.Options, registry *task.TaskRegistry) (*client.TaskHubGrpcWorker, error) {
	return durabletaskscheduler.NewWorker(
		options, registry, api.DefaultLogger(), client.WithAutoWorkItemFilters())
}

Uruchom i opróżnij zwrócony proces roboczy, jak pokazano w przewodniku Go quickstart. Rejestruj tylko zadania, które ten pracownik powinien wykonać.

Zagadnienia do rozważenia

• Ścisłe trasowanie nie ma żadnych zabezpieczeń. Jeśli żaden połączony pracownik nie rejestruje zadania, zadania pozostają w oczekiwaniu, a zależna orkiestracja pozostaje w stanie Running . Upewnij się, że przynajmniej jeden pracownik hostuje każde zadanie wywołane przez workflow.
• Niezgodności rejestracji podczas aktualizacji. Gdy dzielisz zadania lub zmieniasz ich nazwy, wdrażaj procesy robocze obsługujące nowe zadania, zanim zaplanujesz orkiestracje, które od nich zależą.
• Monitoruj zaległości. Użyj panelu monitorowania , aby obserwować oczekujące obowiązki świadczące o braku lub niezdrowym pracowniku.
• Autoscaling. Gdy hostujesz pracowników w Azure Container Apps, reguła skalowania KEDA (Kubernetes Event-driven Autoscaling) może skalować każdego pracownika na podstawie zaległości zarejestrowanych typów zadań. Więcej informacji można znaleźć w artykule Configure autoscale for Azure Container Apps.

Następne kroki

Szybki start: Użyj filtrowania zadań z SDK Durable Task

• Dostęp oparty na tożsamości dla Trwałego Planisty Zadań
• Konfiguruj prywatne punkty końcowe dla Durable Task Scheduler
• Konfigurowanie automatycznego skalowania dla usługi Azure Container Apps