Quickstart: Verwenden Sie Arbeitsitem-Filterung mit den Durable Task SDKs

Voraussetzungen

Bevor Sie beginnen:

• Installieren Sie .NET 10 SDK oder höher.
• Installieren Sie Docker zum Ausführen des Emulators.
• Installieren Sie die Azure Developer CLI.
• Klonen Sie das GitHub-Repository "Durable Task Scheduler".

Vorbereiten des Projekts

Navigieren Sie aus dem Azure-Samples/Durable-Task-Scheduler Stammverzeichnis zum Beispielverzeichnis:

cd samples/scenarios/WorkItemFilteringSplitActivities

Lokal mit dem Emulator ausführen

1. Laden Sie das Emulator-Image herunter:

   docker pull mcr.microsoft.com/durable-task/emulator:latest
   

2. Führen Sie den Emulator aus:

   docker run -d --name dts-emulator -p 8080:8080 -p 8082:8082 mcr.microsoft.com/durable-task/emulator:latest
   

3. Erstellen Sie das Beispiel:

   dotnet build
   

4. Führe jeden Worker in einem separaten Terminal aus:

   Terminal 1 – Orchestrator-Worker

   dotnet run --project src/OrchestratorWorker
   

   Terminal 2 – Validator-Arbeiter

   dotnet run --project src/ValidatorWorker
   

   Terminal 3 – Versandarbeiter

   dotnet run --project src/ShipperWorker
   

5. Führen Sie in einem vierten Terminal den Client aus:

   dotnet run --project src/Client
   

6. Öffnen Sie das Emulatordashboard unter http://localhost:8082, um die Orchestrierungsaktivität zu überwachen.

./run-local.sh

Erwartetes Ergebnis der Arbeitsvorgangsfilterung

Der Client startet alle 30 Sekunden eine Gruppe von drei Orchestrierungen und wiederholt dies etwa 10 Minuten lang. Über die Terminals hinweg sehen Sie:

• Der Client plant Orchestrierungs-Batches und wartet, bis jeder Batch abgeschlossen ist. Ein fertiger Batch protokolliert 3 completed, 0 failed.
• Der Orchestrator-Worker setzt ValidateOrder- und ShipOrder-Aktivitätsaufrufe ab.
• Der Validator-Worker bearbeitet nur ValidateOrder-Aktivitätsarbeitselemente.
• Der Versender-Arbeiter übernimmt nur ShipOrder Aktivitätsaufgaben.

Dieses Verhalten bestätigt, dass die Filterung von Arbeitsaufgaben Elemente nur an Mitarbeiter weitergibt, die übereinstimmende Aufgabentypen registriert haben.

Wenn man mindestens einen Batch-Bericht 3 completed, 0 failedsieht, funktioniert die Probe wie erwartet. Du musst nicht warten, bis die vollständige Schleife vorbei ist – drücke in jedem Terminal Strg+C, um den Client und die Arbeiter zu stoppen. Wenn du den Client bis zum Ende laufen lässt, druckt er eine Endergebniszusammenfassung und bleibt dann lebendig, bis du ihn mit Strg+C stoppst.

Probieren Sie es aus: Striktes Routing-Experiment

Um festzustellen, dass das Routing streng ist (kein Fallback für andere Mitarbeiter):

1. Beenden Sie den Versandarbeiter (STRG+C in Terminal 3).
2. Führen Sie den Client erneut aus, um neue Orchestrierungen zu planen.
3. Beachten Sie Folgendes:
   • Der Orchestrator-Worker nimmt Orchestrierungen auf und startet sie.
   • Der Validator-Worker führt ValidateOrder für jeden Auftrag aus.
   • ShipOrder Arbeitselemente bleiben ausstehend – die Validator- und Orchestrator-Worker empfangen sie nicht.
   • Die Orchestrierungen bleiben im Status „Wird ausgeführt“ und warten, bis ShipOrder abgeschlossen ist.
4. Starte den Shipper Worker neu – die ausstehenden ShipOrder Arbeitsaufgaben werden sofort geliefert und die Orchestrierungen abgeschlossen.

Dieses Experiment zeigt, dass Arbeitsobjekte nur an Arbeiter mit passenden Filtern weitergeleitet werden. Es gibt keine Ausweichmöglichkeit.

Bereitstellen mithilfe der Azure Developer CLI

1. Authentifizieren mit Azure (sofern noch nicht geschehen):

   azd auth login
   

2. Von samples/scenarios/WorkItemFilteringSplitActivities aus ausführen:

   azd up
   

3. Wenn Sie im Terminal dazu aufgefordert werden, geben Sie die folgenden Parameter an:

   Parameter	Beschreibung
   Umgebungsname	Präfix für die Ressourcengruppe, die erstellt wurde, um alle Azure Ressourcen zu enthalten.
   Azure-Standort	Der Azure-Speicherort für Ihre Ressourcen.
   Azure-Abonnement	Das Azure-Abonnement für Ihre Ressourcen.

Der azd up Befehl stellt Azure-Ressourcen bereit und stellt vier containerisierte Dienste aus diesem Beispiel bereit: Client-, Orchestrator-Worker, Validator-Worker und Versandmitarbeiter.

Erfolgreiche Bereitstellung bestätigen

1. Kopieren Sie in der azd up Ausgabe den Ressourcengruppennamen.

2. Öffnen Sie im Azure-Portal diese Ressourcengruppe.

3. Öffnen Sie für jede bereitgestellte Container-App (client, orchestrator-worker, validator-workershipper-worker), denÜberwachungsprotokolldatenstrom>.

4. Überprüfen Sie, ob jede App nur die erwarteten Arbeitsaufgaben protokolliert:

   • orchestrator-worker: Orchestrierungsarbeit.
   • validator-worker: ValidateOrder Aktivität.
   • shipper-worker: ShipOrder Aktivität.

Verstehe den Arbeitsitem-Filtercode in C #

Die Orchestrierung ruft zwei Aktivitäten in Folge auf. Der Scheduler leitet jedes Aktivitätselement an den Arbeiter weiter, der es registriert hat.

public override async Task<string> RunAsync(TaskOrchestrationContext context, string orderId)
{
    string validationResult = await context.CallActivityAsync<string>("ValidateOrder", orderId);
    string shippingResult = await context.CallActivityAsync<string>("ShipOrder", orderId);

    return $"Order '{orderId}' => Validation: [{validationResult}], Shipping: [{shippingResult}]";
}

Jeder Mitarbeiter registriert nur seine lokalen Aufgaben. Das SDK generiert automatisch Arbeitsitem-Filter aus der Task-Registry, sodass du dich nicht explizit anmelden musst.

builder.Services.AddDurableTaskWorker()
    .AddTasks(registry =>
    {
        // Only this worker's tasks are registered.
        // Work item filters are auto-generated from the registry, so this
        // worker only receives matching work items.
        registry.AddAllGeneratedTasks();
    })
    .UseDurableTaskScheduler(connectionString);

Beispiel:

• OrchestratorWorker registriert OrderProcessingOrchestration.
• ValidatorWorker registriert ValidateOrder.
• ShipperWorker registriert ShipOrder.

Wenn ein Worker eine Verbindung mit dem permanenten Aufgabenplaner herstellt, sendet das SDK seine Filterliste. Der Scheduler erstellt Warteschlangen pro Filter und leitet jede Arbeitsaufgabe an die entsprechende Warteschlange weiter. Mitarbeiter erhalten nie Arbeitsaufgabentypen, die sie nicht registriert haben.

Bereinigen von Ressourcen

1. Stoppe den lokalen Emulator-Container.

   docker rm -f dts-emulator
   

2. Wenn Sie in Azure bereitgestellt haben, identifizieren Sie den Namen der Ressourcengruppe:

   • Kopieren Sie es aus der azd up Ausgabe.
   • Oder benutze im Azure-Portal das globale Suchfeld am oberen Rand und suche nach rg- oder dem Präfix deines Umgebungsnamens.

3. Öffnen Sie die Ressourcengruppe aus den Suchergebnissen.

4. Wählen Sie "Ressourcengruppe löschen", geben Sie den zu bestätigenden Namen der Ressourcengruppe ein, und wählen Sie dann "Löschen" aus.

Nächste Schritte

• Erfahren Sie mehr über die automatische Skalierung des permanenten Aufgabenplanungsmoduls in Azure-Container-Apps.
• Lesen Sie anleitungen zur Problembehandlung.

Voraussetzungen

Bevor Sie beginnen:

• Java 21 oder höher.
• Docker zum Ausführen des Emulators.
• Azure Developer CLI.
• Klonen Sie das GitHub-Repository "Durable Task Scheduler".

Vorbereiten des Projekts

Vom Azure-Samples/Durable-Task-Scheduler Root-Verzeichnis aus gehen Sie zum Beispielverzeichnis:

cd samples/scenarios/WorkItemFilteringSplitActivitiesJava

Lokal mit dem Emulator ausführen

1. Zieh das Emulator-Image herunter.

   docker pull mcr.microsoft.com/dts/dts-emulator:latest
   

2. Führen Sie den Emulator aus.

   docker run -d --name dts-emulator -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
   

3. Erstellen Sie das Beispiel.

   ./gradlew build
   

4. Führe jeden Worker in einem separaten Terminal aus:

   Terminal 1 – Orchestrator-Worker

   ./gradlew :orchestrator-worker:run
   

   Terminal 2 – Validator-Arbeiter

   ./gradlew :validator-worker:run
   

   Terminal 3 – Versandarbeiter

   ./gradlew :shipper-worker:run
   

5. Führen Sie den Client in einem vierten Terminal aus.

   ./gradlew :client:run
   

6. Öffnen Sie das Emulatordashboard unter http://localhost:8082, um die Orchestrierungsaktivität zu überwachen.

Erwartetes Ergebnis der Arbeitsvorgangsfilterung

Der Client startet alle 30 Sekunden eine Gruppe von drei Orchestrierungen und wiederholt dies etwa 10 Minuten lang. Über die Terminals hinweg sehen Sie:

• Der Client plant Orchestrierungs-Batches und wartet, bis jeder Batch abgeschlossen ist. Ein fertiger Batch protokolliert 3 completed, 0 failed.
• Der Orchestrator-Worker setzt ValidateOrder- und ShipOrder-Aktivitätsaufrufe ab.
• Der Validator-Worker bearbeitet nur ValidateOrder-Aktivitätsarbeitselemente.
• Der Versender-Arbeiter übernimmt nur ShipOrder Aktivitätsaufgaben.

Dieses Verhalten bestätigt, dass die Filterung von Arbeitsaufgaben Elemente nur an Mitarbeiter weitergibt, die übereinstimmende Aufgabentypen registriert haben.

Wenn man mindestens einen Batch-Bericht 3 completed, 0 failedsieht, funktioniert die Probe wie erwartet. Du musst nicht warten, bis die vollständige Schleife vorbei ist – drücke in jedem Terminal Strg+C, um den Client und die Arbeiter zu stoppen. Wenn du den Client bis zum Ende laufen lässt, druckt er eine Endergebniszusammenfassung und bleibt dann lebendig, bis du ihn mit Strg+C stoppst.

Probieren Sie es aus: Striktes Routing-Experiment

Um festzustellen, dass das Routing streng ist (kein Fallback für andere Mitarbeiter):

1. Beenden Sie den Versandarbeiter (STRG+C in Terminal 3).
2. Führen Sie den Client erneut aus, um neue Orchestrierungen zu planen.
3. Beachten Sie Folgendes:
   • Der Orchestrator-Worker nimmt Orchestrierungen auf und startet sie.
   • Der Validator-Worker führt ValidateOrder für jeden Auftrag aus.
   • ShipOrder Arbeitselemente bleiben ausstehend – die Validator- und Orchestrator-Worker empfangen sie nicht.
   • Die Orchestrierungen bleiben im Status „Wird ausgeführt“ und warten, bis ShipOrder abgeschlossen ist.
4. Starte den Shipper Worker neu – die ausstehenden ShipOrder Arbeitsaufgaben werden sofort geliefert und die Orchestrierungen abgeschlossen.

Dieses Experiment zeigt, dass Arbeitsobjekte nur an Arbeiter mit passenden Filtern weitergeleitet werden. Es gibt keine Ausweichmöglichkeit.

Bereitstellen mithilfe der Azure Developer CLI

1. Authentifizieren mit Azure (sofern noch nicht geschehen):

   azd auth login
   

2. Von samples/scenarios/WorkItemFilteringSplitActivitiesJava aus ausführen:

   azd up
   

3. Wenn Sie im Terminal dazu aufgefordert werden, geben Sie die folgenden Parameter an:

   Parameter	Beschreibung
   Umgebungsname	Präfix für die Ressourcengruppe, die erstellt wurde, um alle Azure Ressourcen zu enthalten.
   Azure-Standort	Der Azure-Speicherort für Ihre Ressourcen.
   Azure-Abonnement	Das Azure-Abonnement für Ihre Ressourcen.

Der azd up Befehl stellt Azure-Ressourcen bereit und stellt vier containerisierte Dienste aus diesem Beispiel bereit: Client-, Orchestrator-Worker, Validator-Worker und Versandmitarbeiter.

Erfolgreiche Bereitstellung bestätigen

1. Kopieren Sie in der azd up Ausgabe den Ressourcengruppennamen.

2. Öffnen Sie im Azure-Portal diese Ressourcengruppe.

3. Öffnen Sie für jede bereitgestellte Container-App (client, orchestrator-worker, validator-workershipper-worker), denÜberwachungsprotokolldatenstrom>.

4. Überprüfen Sie, ob jede App nur die erwarteten Arbeitsaufgaben protokolliert:

   • orchestrator-worker: Orchestrierungsarbeit.
   • validator-worker: ValidateOrder Aktivität.
   • shipper-worker: ShipOrder Aktivität.

Verstehen Sie den Code zur Filterung von Arbeitsgegenständen in Java

Die Orchestrierung ruft zwei Aktivitäten in Folge auf. Der Scheduler leitet jedes Aktivitätselement an den Arbeiter weiter, der es registriert hat.

String validationResult = ctx.callActivity("ValidateOrder", orderId, String.class).await();
String shippingResult = ctx.callActivity("ShipOrder", orderId, String.class).await();

String combined = String.format("Order '%s' => Validation: [%s], Shipping: [%s]",
        orderId, validationResult, shippingResult);
ctx.complete(combined);

Jeder Mitarbeiter registriert nur seine lokalen Aufgaben und Aufrufe useWorkItemFilters() , um die Filterung zu aktivieren. Anschließend generiert das SDK Arbeitsaufgabenfilter aus der Aufgabenregistrierung.

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

Beispiel:

• OrchestratorWorker registriert OrderProcessingOrchestration.
• ValidatorWorker registriert ValidateOrder.
• ShipperWorker registriert ShipOrder.

Wenn ein Worker eine Verbindung mit dem permanenten Aufgabenplaner herstellt, sendet das SDK seine Filterliste. Der Scheduler erstellt Warteschlangen pro Filter und leitet jede Arbeitsaufgabe an die entsprechende Warteschlange weiter. Mitarbeiter erhalten nie Arbeitsaufgabentypen, die sie nicht registriert haben.

Um explizite Filter anstelle der automatischen Generierung aus dem Register bereitzustellen, übergebe a WorkItemFilter an useWorkItemFilters():

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

builder.useWorkItemFilters(filter);

Bereinigen von Ressourcen

1. Stoppe den lokalen Emulator-Container.

   docker rm -f dts-emulator
   

2. Wenn Sie in Azure bereitgestellt haben, identifizieren Sie den Namen der Ressourcengruppe:

   • Kopieren Sie es aus der azd up Ausgabe.
   • Oder benutze im Azure-Portal das globale Suchfeld am oberen Rand und suche nach rg- oder dem Präfix deines Umgebungsnamens.

3. Öffnen Sie die Ressourcengruppe aus den Suchergebnissen.

4. Wählen Sie "Ressourcengruppe löschen", geben Sie den zu bestätigenden Namen der Ressourcengruppe ein, und wählen Sie dann "Löschen" aus.

Nächste Schritte

• Erfahren Sie mehr über die automatische Skalierung des permanenten Aufgabenplanungsmoduls in Azure-Container-Apps.
• Lesen Sie anleitungen zur Problembehandlung.

Voraussetzungen

Bevor Sie beginnen:

• Python 3.10 oder höher.
• Docker zum Ausführen des Emulators.
• Azure CLI, wenn du einen bereitgestellten Durable Task Scheduler verwendest.
• Klonen Sie das GitHub-Repository "Durable Task Scheduler".

Vorbereiten des Projekts

Vom Azure-Samples/Durable-Task-Scheduler Root-Verzeichnis aus gehen Sie zum Beispielverzeichnis:

cd samples/durable-task-sdks/python/work-item-filtering

Lokal mit dem Emulator ausführen

1. Zieh das Emulator-Image herunter.

   docker pull mcr.microsoft.com/dts/dts-emulator:latest
   

2. Führen Sie den Emulator aus.

   docker run -d --name dts-emulator -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
   

   Das Beispiel verwendet die Standard-Emulator-Einstellungen (Endpunkt http://localhost:8080 und Task-Hub default), sodass du keine Umgebungsvariablen setzen musst.

3. (Optional) Erstellen und aktivieren Sie eine virtuelle Python-Umgebung.

   python -m venv venv
   source venv/bin/activate  # On Windows, use: venv\Scripts\activate
   

4. Installieren Sie die erforderlichen Pakete.

   pip install -r requirements.txt
   

5. Führe jeden Worker in einem separaten Terminal aus:

   Terminal 1 – Arbeiter A (Begrüßung)

   python worker_a.py
   

   Terminal 2 – Arbeiter B (Mathematik)

   python worker_b.py
   

6. In einem dritten Terminal starten Sie den Client.

   python client.py
   

7. Öffnen Sie das Emulatordashboard unter http://localhost:8082, um die Orchestrierungsaktivität zu überwachen.

Erwartetes Ergebnis der Arbeitsvorgangsfilterung

Der Client plant eine Begrüßungsorchestrierung und eine mathematische Orchestrierung und wartet dann, bis beide fertig sind. Über die Terminals hinweg sehen Sie:

• Arbeiter A verarbeitet nur die Begrüßungsorchestrierung und ihre zugehörige say_hello-Aktivität.
• Arbeiter B verarbeitet nur die mathematische Orchestrierung und deren add_numbers Aktivität.
• Der Client druckt das Ergebnis jeder Orchestrierung.

Dieses Verhalten bestätigt, dass die Filterung von Arbeitsaufgaben Elemente nur an Mitarbeiter weitergibt, die übereinstimmende Aufgabentypen registriert haben.

Verstehe den Code zum Filtern von Arbeitsobjekten in Python

Jeder Mitarbeiter registriert nur seine lokalen Aufgaben und Aufrufe use_work_item_filters() , um die Filterung zu aktivieren. Das SDK generiert die Arbeitsitem-Filter aus den registrierten Aufgaben.

worker.add_orchestrator(greeting_orchestrator)
worker.add_activity(say_hello)
worker.use_work_item_filters()  # Auto-generate from the registry

Beispiel:

• Arbeiter A registriert greeting_orchestrator und say_hello.
• Arbeiter B registriert math_orchestrator und add_numbers.

Wenn ein Worker eine Verbindung mit dem permanenten Aufgabenplaner herstellt, sendet das SDK seine Filterliste. Der Scheduler erstellt Warteschlangen pro Filter und leitet jede Arbeitsaufgabe an die entsprechende Warteschlange weiter. Mitarbeiter erhalten nie Arbeitsaufgabentypen, die sie nicht registriert haben.

Um explizite Filter bereitzustellen, anstatt sie automatisch aus dem Register zu generieren, gib WorkItemFilters:

from durabletask.worker import WorkItemFilters, OrchestrationWorkItemFilter, ActivityWorkItemFilter

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

Bereinigen von Ressourcen

Stoppe den lokalen Emulator-Container.

docker rm -f dts-emulator

Nächste Schritte

• Erfahren Sie mehr über die automatische Skalierung des permanenten Aufgabenplanungsmoduls in Azure-Container-Apps.
• Lesen Sie anleitungen zur Problembehandlung.