クイックスタート: Durable Task SDK と Durable Task Scheduler を使用してアプリを作成する

Durable Task SDK には、Durable Task Scheduler 用の軽量クライアントライブラリが用意されています。このクイックスタートでは、ファンアウト/ファンインアプリケーションパターンを使用して複数の作業項目を並列処理し、結果を集計するサンプルアプリを作成します。

このクイックスタートの最後には、永続タスクスケジューラーのエミュレーターを使用し、ローカル環境で実行するタスクを並列ワーカーに分散し、結果を集約する実際に機能するオーケストレーションを構築することができます。

Important

現在、PowerShell Durable Task SDK は使用できません。

Note

Durable Task SDK の PowerShell サポートはプレビュー段階です。完全なクイックスタート体験を望むなら、上の別の言語タブ(C#、Python、Java、JavaScript、またはGo)を選択してください。

ローカル開発用の Durable Task Scheduler エミュレーターを設定して実行します。
ワーカーとクライアントの両方を実行しましょう。
Durable Task Scheduler ダッシュボードを使用して、オーケストレーションの状態と履歴を確認します。

前提条件

開始する前に、次の手順を実行します。

.NET 10 SDK 以降があることを確認します。
エミュレーターを実行するための Docker をインストールします。
クイックスタートサンプルを使用するには、Durable Task Scheduler GitHub リポジトリを複製します。

Python 3.9 以降がインストールされていることを確認してください。
エミュレーターを実行するための Docker をインストールします。
クイックスタートサンプルを使用するには、Durable Task Scheduler GitHub リポジトリを複製します。

Java 8 または 11 があることを確認します。
エミュレーターを実行するための Docker をインストールします。
クイックスタートサンプルを使用するには、Durable Task Scheduler GitHub リポジトリを複製します。

Node.js 22 以降があることを確認します。
エミュレーターを実行するための Docker をインストールします。
クイックスタートサンプルを使用するには、Durable Task Scheduler GitHub リポジトリを複製します。

Go 1.25以降をインストールしてください。
Dockerまたは対応するコンテナランタイムをインストールしてください。

Note

Go SDKはベータ版です。このクイックスタートは github.com/microsoft/durabletask-go@v1.0.0-beta3 を使い、新しいセルフホスト型 Go アプリを作成します。 Azureのサブスクリプションやサンプルリポジトリのチェックアウトは必要ありません。

Durable Task Scheduler エミュレーターを設定する

エミュレーターは、Docker コンテナー内の Durable Task Scheduler とタスクハブをシミュレートするため、ローカル開発に最適です。

このサンプルコードは、環境変数が設定されていない場合に、既定でエミュレーターに接続します。

Azure-Samples/Durable-Task-Scheduler ルートディレクトリから、.NET SDK サンプルディレクトリに移動します。
```
cd samples/durable-task-sdks/dotnet/FanOutFanIn
```
エミュレーターの Docker イメージをプルします。
```
docker pull mcr.microsoft.com/dts/dts-emulator:latest
```
エミュレーターを実行します。コンテナーの準備に数秒かかる場合があります。
```
docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
```

サンプルコードでは既定のエミュレーター設定が自動的に使用されるため、環境変数を設定する必要はありません。このクイックスタートの既定のエミュレーター設定は次のとおりです。

エンドポイント: http://localhost:8080
タスクハブ: default

Azure-Samples/Durable-Task-Schedulerルートディレクトリから、Python SDK サンプルディレクトリに移動します。
```
cd samples/durable-task-sdks/python/fan-out-fan-in
```
エミュレーターの Docker イメージをプルします。
```
docker pull mcr.microsoft.com/dts/dts-emulator:latest
```
エミュレーターを実行します。コンテナーの準備に数秒かかる場合があります。
```
docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
```

エンドポイント: http://localhost:8080
タスクハブ: default

Azure-Samples/Durable-Task-Schedulerルートディレクトリから、Java SDK サンプルディレクトリに移動します。
```
cd samples/durable-task-sdks/java/fan-out-fan-in
```
エミュレーターの Docker イメージをプルします。
```
docker pull mcr.microsoft.com/dts/dts-emulator:latest
```
エミュレーターを実行します。コンテナーの準備に数秒かかる場合があります。
```
docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
```

エンドポイント: http://localhost:8080
タスクハブ: default

Azure-Samples/Durable-Task-Scheduler ルートディレクトリから、JavaScript SDK サンプルディレクトリに移動します。
```
cd samples/durable-task-sdks/javascript/fan-out-fan-in
```
エミュレーターの Docker イメージをプルします。
```
docker pull mcr.microsoft.com/dts/dts-emulator:latest
```
エミュレーターを実行します。コンテナーの準備に数秒かかる場合があります。
```
docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
```

エンドポイント: http://localhost:8080
タスクハブ: default

専用エミュレーターを起動する。ポートをループバックにバインドして、認証されていないエミュレーターを自分のマシンにローカルに保ちましょう。
```
docker run --name go-dts-emulator -d \
  -p 127.0.0.1:8080:8080 -p 127.0.0.1:8082:8082 \
  mcr.microsoft.com/dts/dts-emulator:latest
```
http://localhost:8082のダッシュボードが利用可能になるのを待ちましょう。もし他のエミュレーターがすでにこれらのポートを使っているなら、そのエミュレーターを使うか、2つ目のエミュレーターを起動する前に停止してください。

Goアプリを実行するターミナルで接続設定をしてください。

export DTS_CONNECTION_STRING="Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"

Goモジュールを作成し、SDKをインストールします。

mkdir durable-task-go-quickstart
cd durable-task-go-quickstart
go mod init example.com/durable-task-go-quickstart
go get github.com/microsoft/durabletask-go@v1.0.0-beta3

クイックスタートを実行する

FanOutFanIn ディレクトリから、Worker ディレクトリに移動してワーカーをビルドして実行します。
```
cd Worker
dotnet build
dotnet run
```
別のターミナルで、 FanOutFanIn ディレクトリから Client ディレクトリに移動して、クライアントをビルドして実行します。
```
cd Client
dotnet build
dotnet run
```

出力を理解すること

このサンプルを実行すると、ワーカープロセスとクライアントプロセスの両方から出力を受け取ります。プロジェクトの実行時にコードで何が起こったかをアンパックします。

作業者の成果

ワーカーの出力には、次の情報が表示されます。

オーケストレーターとアクティビティの登録
各アクティビティが呼び出されたときにエントリをログに記録する
複数の作業項目の並列処理
結果の最終的な集計

クライアント出力

クライアントの出力は次のように表示されます。

オーケストレーションは作業項目の一覧から始まります。
一意のオーケストレーションインスタンス ID
各作業項目とそれに対応する結果を示す最終的な集計結果
処理済みアイテムの合計数

サンプルの出力

Starting Fan-Out Fan-In Pattern - Parallel Processing Client
Using local emulator with no authentication
Starting parallel processing orchestration with 5 work items
Work items: ["Task1","Task2","Task3","LongerTask4","VeryLongTask5"]
Started orchestration with ID: 7f8e9a6b-1c2d-3e4f-5a6b-7c8d9e0f1a2b
Waiting for orchestration to complete...
Orchestration completed with status: Completed
Processing results:
Work item: Task1, Result: 5
Work item: Task2, Result: 5
Work item: Task3, Result: 5
Work item: LongerTask4, Result: 11
Work item: VeryLongTask5, Result: 13
Total items processed: 5

Python 仮想環境をアクティブ化します。

ウィンドウズ
リナックス

python -m venv venv
/venv/Scripts/activate

python -m venv venv
source ./venv/bin/activate

必要なパッケージをインストールします。
```
pip install -r requirements.txt
```

ワーカーを開始します。

python worker.py

想定される出力

ワーカーが起動され、作業項目を待っていることを示す出力を確認できます。

Starting Fan Out/Fan In pattern worker...
Using taskhub: default
Using endpoint: http://localhost:8080
Starting gRPC worker that connects to http://localhost:8080
Successfully connected to http://localhost:8080. Waiting for work items...

新しいターミナルで、仮想環境をアクティブ化し、クライアントを実行します。
- ウィンドウズ
- リナックス
```
venv/Scripts/activate
python client.py
```
```
source ./venv/bin/activate
python client.py
```
作業項目の数を引数として指定できます。引数が指定されていない場合、この例では既定で 10 個の項目が実行されます。
```
python client.py [number_of_items]
```

出力を理解すること

作業者の成果

ワーカーの出力には、次の情報が表示されます。

オーケストレーターとアクティビティの登録。
各作業項目を並行して処理するときに、同時に実行されていることを示すステータスメッセージ。
さまざまな処理時間をシミュレートするための各作業項目のランダムな遅延 (0.5 ~ 2 秒)。
結果の集計を示す最後のメッセージ。

クライアント出力

クライアントの出力は次のように表示されます。

オーケストレーションは指定された作業項目の数から始まります。
一意のオーケストレーションインスタンス ID。
最終的な集計結果には、次のものが含まれます。
- 処理されたアイテムの合計数
- すべての結果の合計 (各項目の結果は値の 2 乗)
- すべての結果の平均

サンプルの出力

Starting fan out/fan in orchestration with 10 items
Waiting for 10 parallel tasks to complete
Orchestrator yielded with 10 task(s) and 0 event(s) outstanding.
Processing work item: 1
Processing work item: 2
Processing work item: 10
Processing work item: 9
Processing work item: 8
Processing work item: 7
Processing work item: 6
Processing work item: 5
Processing work item: 4
Processing work item: 3
Orchestrator yielded with 9 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 8 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 7 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 6 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 5 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 4 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 3 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 2 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 1 task(s) and 0 event(s) outstanding.
All parallel tasks completed, aggregating results
Orchestrator yielded with 1 task(s) and 0 event(s) outstanding.
Aggregating results from 10 items
Orchestration completed with status: COMPLETED

fan-out-fan-in ディレクトリから、Gradle を使用してアプリケーションをビルドして実行します。

./gradlew runFanOutFanInPattern

ヒント

zsh: permission denied: ./gradlewエラーメッセージが表示された場合は、アプリケーションを実行する前にchmod +x gradlewを実行してみてください。

出力を理解すること

このサンプルを実行すると、次のような出力が表示されます。

オーケストレーターとアクティビティの登録。
各作業項目を並行して処理するときに、同時に実行されていることを示すステータスメッセージ。
さまざまな処理時間をシミュレートするための各作業項目のランダムな遅延 (0.5 ~ 2 秒)。
結果の集計を示す最後のメッセージ。

プロジェクトの実行時にコードで何が起こったかをアンパックします。

サンプルの出力

Starting a Gradle Daemon (subsequent builds will be faster)

> Task :runFanOutFanInPattern
Durable Task worker is connecting to sidecar at localhost:8080.
Started new orchestration instance
Orchestration completed: [Name: 'FanOutFanIn_WordCount', ID: '<id-number>', RuntimeStatus: COMPLETED, CreatedAt: 2025-04-25T15:24:47.170Z, LastUpdatedAt: 2025-04-25T15:24:47.287Z, Input: '["Hello, world!","The quick brown fox jumps over t...', Output: '60']
Output: 60

必要な npm パッケージをインストールします。
```
npm install
```
ワーカーを開始します。
```
npm run worker
```
想定される出力

ワーカーが起動され、作業項目を待っていることを示す出力を確認できます。
```
Using local emulator with no authentication
Starting worker...
Worker is ready and listening for tasks.
```
新しいターミナルで、 fan-out-fan-in ディレクトリからクライアントを実行します。
```
npm run client
```
作業項目の数を引数として指定できます。引数が指定されていない場合、この例では既定で 10 個の項目が実行されます。
```
npm run client -- 15
```

出力を理解すること

作業者の成果

ワーカーの出力には、次の情報が表示されます。

各作業項目を並行して処理するときに、同時に実行されていることを示すステータスメッセージ。
さまざまな処理時間をシミュレートするための各作業項目のランダムな遅延 (0.5 ~ 2 秒)。
結果の集計を示す最後のメッセージ。

クライアント出力

クライアントの出力は次のように表示されます。

オーケストレーションは指定された作業項目の数から始まります。
一意のオーケストレーションインスタンス ID。
最終的な集計結果には、次のものが含まれます。
- 処理されたアイテムの合計数
- すべての結果の合計 (各項目の結果は値の 2 乗)
- すべての結果の平均

サンプルの出力

Using local emulator with no authentication
Starting fan out/fan in orchestration with 10 items
Started orchestration with ID: abc123-def456-ghi789
Waiting for orchestration to complete...
Processing work item 1 (delay 823ms)
Processing work item 2 (delay 1205ms)
Processing work item 3 (delay 512ms)
Processing work item 4 (delay 1890ms)
Processing work item 5 (delay 645ms)
Processing work item 6 (delay 1102ms)
Processing work item 7 (delay 933ms)
Processing work item 8 (delay 1567ms)
Processing work item 9 (delay 701ms)
Processing work item 10 (delay 1344ms)
Aggregating 10 results...
Orchestration completed with status: COMPLETED
Result: {"totalItems":10,"sum":385,"average":38.5,"results":[...]}

以下の完全なプログラムをmain.go に durable-task-go-quickstart として保存します:

package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"time"

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

type summary struct {
	TotalItems int     `json:"total_items"`
	Sum        int64   `json:"sum"`
	Average    float64 `json:"average"`
}

func fanOutFanIn(ctx *task.OrchestrationContext) (any, error) {
	var items []int64
	if err := ctx.GetInput(&items); err != nil {
		return nil, err
	}
	if len(items) > 100 {
		return nil, fmt.Errorf("batch contains more than 100 items")
	}

	pending := make([]task.Task, len(items))
	for i, item := range items {
		pending[i] = ctx.CallActivity("ProcessWorkItem", task.WithActivityInput(item))
	}
	if err := ctx.WhenAll(pending...); err != nil {
		return nil, fmt.Errorf("process batch: %w", err)
	}

	result := summary{TotalItems: len(items)}
	for _, work := range pending {
		var squared int64
		if err := work.Await(&squared); err != nil {
			return nil, err
		}
		result.Sum += squared
	}
	if result.TotalItems != 0 {
		result.Average = float64(result.Sum) / float64(result.TotalItems)
	}
	return result, nil
}

func processWorkItem(ctx task.ActivityContext) (any, error) {
	var item int64
	if err := ctx.GetInput(&item); err != nil {
		return nil, err
	}
	if item < -1_000_000 || item > 1_000_000 {
		return nil, fmt.Errorf("item %d exceeds the sample's safe arithmetic range", item)
	}
	return item * item, nil
}

func run(ctx context.Context) (err error) {
	options, err := durabletaskscheduler.NewOptionsFromConnectionString(
		os.Getenv("DTS_CONNECTION_STRING"))
	if err != nil {
		return err
	}

	registry := task.NewTaskRegistry()
	if err := registry.AddOrchestratorN("FanOutFanIn", fanOutFanIn); err != nil {
		return err
	}
	if err := registry.AddActivityN("ProcessWorkItem", processWorkItem); err != nil {
		return err
	}

	logger := api.DefaultLogger()
	durableClient, err := durabletaskscheduler.NewClient(ctx, options, logger)
	if err != nil {
		return err
	}
	defer func() {
		err = errors.Join(err, durableClient.Close())
	}()

	worker, err := durabletaskscheduler.NewWorker(
		options, registry, logger, client.WithAutoWorkItemFilters())
	if err != nil {
		return err
	}
	if err := worker.Start(ctx); err != nil {
		return err
	}
	defer func() {
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		err = errors.Join(err, worker.Shutdown(shutdownCtx))
	}()

	id, err := durableClient.ScheduleNewOrchestration(
		ctx, "FanOutFanIn",
		api.WithInput([]int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}))
	if err != nil {
		return err
	}
	metadata, err := durableClient.WaitForOrchestrationCompletion(
		ctx, id, api.WithFetchPayloads(true))
	if err != nil {
		return err
	}
	if metadata.RuntimeStatus != api.RUNTIME_STATUS_COMPLETED {
		return fmt.Errorf("orchestration ended with %s: %+v",
			metadata.RuntimeStatus, metadata.FailureDetails)
	}
	var result summary
	if err := metadata.ReadOutput(&result); err != nil {
		return err
	}
	fmt.Printf("Instance: %s\nStatus: %s\n", id, metadata.RuntimeStatus)
	fmt.Printf("Items: %d\nSum: %d\nAverage: %.1f\n",
		result.TotalItems, result.Sum, result.Average)
	return nil
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	if err := run(ctx); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

プログラムを実行します:

go mod tidy
go run .

コードを保存した後にgo mod tidyを実行して、GoがSDKの推移依存関係やチェックサムを記録できるようにします。

ワーカーとクライアントは同じプロセスで動作します。ワーカーがドレインし、デモンストレーション終了後にクライアントが閉じます。インスタンスはエミュレーターに残っているので、その履歴を確認できます。

出力を理解すること

このプログラムは1から10までの数字を並列に二乗し、結果を集計します。インスタンスIDは様々ですが、最終的な値は以下の通りです:

Instance: <generated-instance-id>
Status: ORCHESTRATION_STATUS_COMPLETED
Items: 10
Sum: 385
Average: 38.5

接続、アクティビティ、復号、シャットダウンエラーは、非ゼロの終了ステータスを生み出します。致命的なワークフローの失敗は成功した完了とはみなされません。

プロジェクトをローカルで実行したので、Azure Container Apps でホストされているAzureにデプロイする方法について説明します。

オーケストレーションの状態と履歴を表示する

Durable Task Scheduler ダッシュボードを使用して、オーケストレーションの状態と履歴を表示できます。既定では、ダッシュボードはポート 8082 で実行されます。

Web ブラウザーで http://localhost:8082 に移動します。
既定のタスクハブをクリックします。作成したオーケストレーションインスタンスが一覧に表示されます。
オーケストレーションインスタンス ID をクリックして、次のような実行の詳細を表示します。
- 複数のアクティビティタスクの並列実行
- ファンイン集計手順
- 各ステップでの入力と出力
- 各ステップにかかった時間

.NET サンプルのオーケストレーションインスタンスの詳細を示すスクリーンショット。

Python サンプルのオーケストレーションインスタンスの詳細を示すスクリーンショット。

Java サンプルのオーケストレーションインスタンスの詳細を示すスクリーンショット

JavaScript サンプルのオーケストレーションインスタンスの詳細を示すスクリーンショット。

コード構造について

労働者プロジェクト

ファンアウト/ファンインパターンを示すために、worker プロジェクトオーケストレーションは並列アクティビティタスクを作成し、すべてが完了するまで待機します。オーケストレーター:

作業項目の一覧を入力として受け取ります。
ProcessWorkItemActivityを使用して作業項目ごとに個別のタスクを作成してファンアウトします。
すべてのタスクを並列で実行します。
Task.WhenAllを使用して、すべてのタスクが完了するまで待機します。
AggregateResultsActivity を使用して、すべての個々の結果を集計してファンインします。
最終的な集計結果をクライアントに返します。

作業員のプロジェクトには次のものが含まれます。

ParallelProcessingOrchestration.cs: オーケストレーター関数とアクティビティ関数を 1 つのファイルで定義します。
Program.cs: 適切な接続文字列処理を使用してワーカーホストを設定します。

`ParallelProcessingOrchestration.cs`

ファンアウト/ファンインを使用して、オーケストレーションは並列アクティビティタスクを作成し、すべてが完了するまで待機します。

public override async Task<Dictionary<string, int>> RunAsync(TaskOrchestrationContext context, List<string> workItems)
{
    // Step 1: Fan-out by creating a task for each work item in parallel
    List<Task<Dictionary<string, int>>> processingTasks = new List<Task<Dictionary<string, int>>>();

    foreach (string workItem in workItems)
    {
        // Create a task for each work item (fan-out)
        Task<Dictionary<string, int>> task = context.CallActivityAsync<Dictionary<string, int>>(
            nameof(ProcessWorkItemActivity), workItem);
        processingTasks.Add(task);
    }

    // Step 2: Wait for all parallel tasks to complete
    Dictionary<string, int>[] results = await Task.WhenAll(processingTasks);

    // Step 3: Fan-in by aggregating all results
    Dictionary<string, int> aggregatedResults = await context.CallActivityAsync<Dictionary<string, int>>(
        nameof(AggregateResultsActivity), results);

    return aggregatedResults;
}

各アクティビティは、 [DurableTask] 属性で修飾された個別のクラスとして実装されます。

[DurableTask]
public class ProcessWorkItemActivity : TaskActivity<string, Dictionary<string, int>>
{
    // Implementation processes a single work item
}

[DurableTask]
public class AggregateResultsActivity : TaskActivity<Dictionary<string, int>[], Dictionary<string, int>>
{
    // Implementation aggregates individual results
}

`Program.cs`

ワーカーは、適切なライフサイクル管理のために Microsoft.Extensions.Hosting を使用します。

using Microsoft.Extensions.Hosting;
//..

builder.Services.AddDurableTaskWorker()
    .AddTasks(registry =>
    {
        registry.AddOrchestrator<ParallelProcessingOrchestration>();
        registry.AddActivity<ProcessWorkItemActivity>();
        registry.AddActivity<AggregateResultsActivity>();
    })
    .UseDurableTaskScheduler(connectionString);

クライアントプロジェクト

クライアントプロジェクト:

ワーカーと同じコネクションストリングロジックを使用します。
並列処理する作業項目の一覧を作成します。
リストを入力として使用してオーケストレーションインスタンスをスケジュールします。
オーケストレーションが完了するまで待機し、集計された結果を表示します。
効率的なポーリングに WaitForInstanceCompletionAsync を使用します。

List<string> workItems = new List<string>
{
    "Task1",
    "Task2",
    "Task3",
    "LongerTask4",
    "VeryLongTask5"
};

// Schedule the orchestration with the work items
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
    "ParallelProcessingOrchestration", 
    workItems);

// Wait for completion
var instance = await client.WaitForInstanceCompletionAsync(
    instanceId,
    getInputsAndOutputs: true,
    cts.Token);

`worker.py`

作業項目の一覧を入力として受け取ります。
作業項目ごとに並列タスクを作成して "ファンアウト" します (各作業項目に対して process_work_item を呼び出します)。
task.when_allを使用して、すべてのタスクが完了するまで待機します。
その後、aggregate_results アクティビティで結果を集計して "ファンイン" します。
最終的な集計結果がクライアントに返されます。

ファンアウト/ファンインを使用して、オーケストレーションは並列アクティビティタスクを作成し、すべてが完了するまで待機します。

# Orchestrator function
def fan_out_fan_in_orchestrator(ctx, work_items: list) -> dict:
    logger.info(f"Starting fan out/fan in orchestration with {len(work_items)} items")

    # Fan out: Create a task for each work item
    parallel_tasks = []
    for item in work_items:
        parallel_tasks.append(ctx.call_activity("process_work_item", input=item))

    # Wait for all tasks to complete
    logger.info(f"Waiting for {len(parallel_tasks)} parallel tasks to complete")
    results = yield task.when_all(parallel_tasks)

    # Fan in: Aggregate all the results
    logger.info("All parallel tasks completed, aggregating results")
    final_result = yield ctx.call_activity("aggregate_results", input=results)

    return final_result

`client.py`

クライアントプロジェクト:

ワーカーと同じコネクションストリングロジックを使用します。
並列処理する作業項目の一覧を作成します。
リストを入力として使用してオーケストレーションインスタンスをスケジュールします。
オーケストレーションが完了するまで待機し、集計された結果を表示します。
効率的なポーリングに wait_for_orchestration_completion を使用します。

# Generate work items (default 10 items if not specified)
count = int(sys.argv[1]) if len(sys.argv) > 1 else 10
work_items = list(range(1, count + 1))

logger.info(f"Starting new fan out/fan in orchestration with {count} work items")

# Schedule a new orchestration instance
instance_id = client.schedule_new_orchestration(
    "fan_out_fan_in_orchestrator", 
    input=work_items
)

logger.info(f"Started orchestration with ID = {instance_id}")

# Wait for orchestration to complete
logger.info("Waiting for orchestration to complete...")
result = client.wait_for_orchestration_completion(
    instance_id,
    timeout=60
)

ファンアウト/ファンインパターンを示すために、FanOutFanInPattern プロジェクトオーケストレーションは並列アクティビティタスクを作成し、すべてが完了するまで待機します。オーケストレーター:

作業項目の一覧を入力として受け取ります。
callActivityを使用して作業項目ごとに個別のタスクを作成してファンアウトします。
すべてのタスクを並列で実行します。
allOfを使用して、すべてのタスクが完了するまで待機します。
stream().mapToInt().sum() を使用して、すべての個々の結果を集計してファンインします。
最終的な集計結果をクライアントに返します。

プロジェクトには次のものが含まれます。

DurableTaskSchedulerWorkerExtensions worker: オーケストレーター関数とアクティビティ関数を定義します。
DurableTaskSchedulerClientExtension クライアント: 適切な接続文字列処理を使用してワーカーホストを設定します。

作業者

ファンアウト/ファンインを使用して、オーケストレーションは並列アクティビティタスクを作成し、すべてが完了するまで待機します。

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions.createWorkerBuilder(connectionString)
    .addOrchestration(new TaskOrchestrationFactory() {
        @Override
        public String getName() { return "FanOutFanIn_WordCount"; }

        @Override
        public TaskOrchestration create() {
            return ctx -> {
                List<?> inputs = ctx.getInput(List.class);
                List<Task<Integer>> tasks = inputs.stream()
                        .map(input -> ctx.callActivity("CountWords", input.toString(), Integer.class))
                        .collect(Collectors.toList());
                List<Integer> allWordCountResults = ctx.allOf(tasks).await();
                int totalWordCount = allWordCountResults.stream().mapToInt(Integer::intValue).sum();
                ctx.complete(totalWordCount);
            };
        }
    })
    .addActivity(new TaskActivityFactory() {
        @Override
        public String getName() { return "CountWords"; }

        @Override
        public TaskActivity create() {
            return ctx -> {
                String input = ctx.getInput(String.class);
                StringTokenizer tokenizer = new StringTokenizer(input);
                return tokenizer.countTokens();
            };
        }
    })
    .build();

// Start the worker
worker.start();

クライアント

クライアントプロジェクト:

ワーカーと同じコネクションストリングロジックを使用します。
並列処理する作業項目の一覧を作成します。
リストを入力として使用してオーケストレーションインスタンスをスケジュールします。
オーケストレーションが完了するまで待機し、集計された結果を表示します。
効率的なポーリングに waitForInstanceCompletion を使用します。

DurableTaskClient client = DurableTaskSchedulerClientExtensions.createClientBuilder(connectionString).build();

// The input is an arbitrary list of strings.
List<String> listOfStrings = Arrays.asList(
        "Hello, world!",
        "The quick brown fox jumps over the lazy dog.",
        "If a tree falls in the forest and there is no one there to hear it, does it make a sound?",
        "The greatest glory in living lies not in never falling, but in rising every time we fall.",
        "Always remember that you are absolutely unique. Just like everyone else.");

// Schedule an orchestration which will reliably count the number of words in all the given sentences.
String instanceId = client.scheduleNewOrchestrationInstance(
        "FanOutFanIn_WordCount",
        new NewOrchestrationInstanceOptions().setInput(listOfStrings));
logger.info("Started new orchestration instance: {}", instanceId);

// Block until the orchestration completes. Then print the final status, which includes the output.
OrchestrationMetadata completedInstance = client.waitForInstanceCompletion(
        instanceId,
        Duration.ofSeconds(30),
        true);
logger.info("Orchestration completed: {}", completedInstance);
logger.info("Output: {}", completedInstance.readOutputAs(int.class));

ファンアウト/ファンインパターンを示すために、サンプルオーケストレーションは並列アクティビティタスクを作成し、すべてが完了するまで待機します。オーケストレーター:

作業項目の一覧を入力として受け取ります。
作業項目ごとに並列タスクを作成してファンアウトします (各作業項目に対して processWorkItem を呼び出します)。
whenAllを使用して、すべてのタスクが完了するまで待機します。
aggregateResults アクティビティを使用して、結果を集約して 1 つにまとめます。
最終的な集計結果をクライアントに返します。

プロジェクトには次のものが含まれます。

worker.mjs: オーケストレーター関数とアクティビティ関数を定義します。
client.mjs: オーケストレーションをスケジュールし、完了を待機します。

`worker.mjs`

ファンアウト/ファンインを使用して、オーケストレーションは並列アクティビティタスクを作成し、すべてが完了するまで待機します。

import { whenAll } from "@microsoft/durabletask-js";
import { createAzureManagedWorkerBuilder } from "@microsoft/durabletask-js-azuremanaged";

// Orchestrator function using generator syntax
const fanOutFanInOrchestrator = async function* (ctx, workItems) {
  const items = Array.isArray(workItems) ? workItems : [];

  // Fan out: schedule parallel activity calls
  const tasks = items.map((item) => ctx.callActivity(processWorkItem, item));

  // Wait for all tasks using whenAll
  const processedResults = yield whenAll(tasks);

  // Fan in: aggregate results
  const finalResult = yield ctx.callActivity(aggregateResults, processedResults);

  return finalResult;
};

// Activity that processes a single work item
const processWorkItem = async (_ctx, workItem) => {
  const normalizedItem = Number(workItem);
  const delayMs = 500 + Math.floor(Math.random() * 1500);
  console.log(`Processing work item ${normalizedItem} (delay ${delayMs}ms)`);

  await new Promise((resolve) => setTimeout(resolve, delayMs));

  return {
    item: normalizedItem,
    result: normalizedItem * normalizedItem,
  };
};

// Activity that aggregates results
const aggregateResults = async (_ctx, results) => {
  const sum = results.reduce((acc, curr) => acc + curr.result, 0);

  return {
    totalItems: results.length,
    sum,
    average: results.length ? sum / results.length : 0,
    results,
  };
};

// Build and start worker
const connectionString = `Endpoint=http://localhost:8080;Authentication=None;TaskHub=default`;
const worker = createAzureManagedWorkerBuilder(connectionString)
  .addOrchestrator(fanOutFanInOrchestrator)
  .addActivity(processWorkItem)
  .addActivity(aggregateResults)
  .build();

await worker.start();

`client.mjs`

クライアントプロジェクト:

ワーカーと同じコネクションストリングロジックを使用します。
並列処理する作業項目の一覧を作成します。
リストを入力として使用してオーケストレーションインスタンスをスケジュールします。
オーケストレーションが完了するまで待機し、集計された結果を表示します。
効率的なポーリングに waitForOrchestrationCompletion を使用します。

import { OrchestrationStatus } from "@microsoft/durabletask-js";
import { createAzureManagedClient } from "@microsoft/durabletask-js-azuremanaged";

// Create work items array
const count = 10;
const workItems = Array.from({ length: count }, (_unused, index) => index + 1);

const connectionString = `Endpoint=http://localhost:8080;Authentication=None;TaskHub=default`;
const client = createAzureManagedClient(connectionString);

// Schedule orchestration
const instanceId = await client.scheduleNewOrchestration(
  "fanOutFanInOrchestrator",
  workItems
);

console.log(`Started orchestration with ID: ${instanceId}`);

// Wait for completion
const state = await client.waitForOrchestrationCompletion(instanceId, true, 120);

if (state.runtimeStatus === OrchestrationStatus.COMPLETED) {
  console.log(`Result: ${state.serializedOutput}`);
}

await client.stop();

fanOutFanIn 待機する前にすべてのアクティビティをスケジュールします。 ctx.WhenAll バッチ全体を待ち、1 つが失敗した場合の残りのタスクを含めて待機します。 Await 次に各アクティビティの結果をデコードします。このアプローチは永続的な並行処理を提供します。オーケストレーターはネイティブのゴルーチンを作成しません。

processWorkItem オーケストレーターの外で動作し、通常のGo I/Oを実行できます。このサンプルは代わりに算術を用い、オーバーフローを避けるために入力範囲を制限します。また、オーケストレーターはバッチサイズにも制限を設けています。

run タスク名を登録し、クライアントとワーカーの別接続を作成し、ワーカーを起動し、1つのインスタンスをスケジューリングします。出力ペイロードを要求し、RuntimeStatusをチェックし、型付き出力を読み取ります。遅延シャットダウンは新しいコンテキストを使うため、メインのタイムアウト後もクリーンアップが実行可能です。

その他のワークフローパターンやベータ版の制限については、Go 用 Durable Task SDK を使用して開発するをご覧ください。

リソースをクリーンアップする

完了したら、エミュレーターコンテナーを停止します。

docker stop dtsemulator && docker rm dtsemulator

このクイックスタートのために go-dts-emulator を開始した場合は、それを停止して削除してください:

docker stop go-dts-emulator && docker rm go-dts-emulator

次のステップ

Durable Task Scheduler エミュレーターを使用してサンプルをローカルで実行したので、スケジューラとタスクハブリソースを作成し、Azure Container Apps にデプロイしてみてください。

Azure Container Apps でホストされている Durable Task Scheduler を Azure にデプロイする

Durable Task SDK for Go で開発

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-10-15

クイック スタート: Durable Task SDK と Durable Task Scheduler を使用してアプリを作成する

前提条件

Durable Task Scheduler エミュレーターを設定する

クイック スタートを実行する

出力を理解すること

作業者の成果

クライアント出力

サンプルの出力

出力を理解すること

作業者の成果

クライアント出力

サンプルの出力

出力を理解すること

サンプルの出力

出力を理解すること

作業者の成果

クライアント出力

サンプルの出力

出力を理解すること

オーケストレーションの状態と履歴を表示する

コード構造について

労働者プロジェクト

ParallelProcessingOrchestration.cs

Program.cs

クライアント プロジェクト

worker.py

client.py

作業者

クライアント

worker.mjs

client.mjs

リソースをクリーンアップする

次のステップ

フィードバック

その他のリソース

クイックスタート: Durable Task SDK と Durable Task Scheduler を使用してアプリを作成する

クイックスタートを実行する

`ParallelProcessingOrchestration.cs`

`Program.cs`

クライアントプロジェクト

`worker.py`

`client.py`

`worker.mjs`

`client.mjs`