Queue サービスを作成する

[アーティクル]
12/13/2023

Queue サービスは、実行時間の長いサービスの良い例です。作業項目をキューに入れ、前の作業項目が完了してから順番に処理できます。 Worker サービステンプレートに依存して、BackgroundService の上に新しい機能を構築します。

このチュートリアルでは、次の作業を行う方法について説明します。

Queue サービスを作成します。
タスクキューに作業を委任します。
IHostApplicationLifetime イベントからコンソールキーリスナーを登録します。

ヒント

".NET でのワーカー" のサンプルソースコードはすべて、サンプルブラウザーでダウンロードできます。詳細については、コードサンプルの参照: .NET でのワーカーに関するページをご覧ください。

前提条件

.NET 8.0 SDK 以降
.NET 統合開発環境 (IDE)
- Visual Studio を自由に使用できます

新しいプロジェクトを作る

Visual Studio を使用して新しい Worker サービスプロジェクトを作成するには、[ファイル]>[新規]>[プロジェクト] を選択します。[新しいプロジェクトの作成] ダイアログで "Worker サービス" を検索し、Worker サービステンプレートを選択します。 .NET CLI を使用する場合は、作業ディレクトリで好みのターミナルを開きます。 dotnet new コマンドを実行し、<Project.Name> を目的のプロジェクト名に置き換えます。

dotnet new worker --name <Project.Name>

.NET CLI の new worker サービスプロジェクトコマンドの詳細については、「dotnet new worker」を参照してください。

ヒント

Visual Studio Code を使用している場合は、統合ターミナルから .NET CLI コマンドを実行できます。詳細については、Visual Studio Code の統合ターミナルに関する記事を参照してください。

キューサービスを作成する

System.Web.Hosting 名前空間の QueueBackgroundWorkItem(Func<CancellationToken,Task>) 機能についてよく理解しているかもしれません。

ヒント

System.Web 名前空間の機能は意図的に .NET に移植されず、.NET Framework 専用のままです。詳細については、「増分 ASP.NET から ASP.NET Core への移行の概要」を参照してください。

.NET では、この QueueBackgroundWorkItem 機能にインスピレーションを得てサービスをモデル化するには、まず、IBackgroundTaskQueue インターフェイスをプロジェクトに追加します:

namespace App.QueueService;

public interface IBackgroundTaskQueue
{
    ValueTask QueueBackgroundWorkItemAsync(
        Func<CancellationToken, ValueTask> workItem);

    ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(
        CancellationToken cancellationToken);
}

2 つのメソッドがあります。1 つはキュー機能を公開し、もう 1 つは以前にキューに入れられた作業項目をデキューします。 "作業項目" は Func<CancellationToken, ValueTask> です。次に、既定の実装をプロジェクトに追加します。

using System.Threading.Channels;

namespace App.QueueService;

public sealed class DefaultBackgroundTaskQueue : IBackgroundTaskQueue
{
    private readonly Channel<Func<CancellationToken, ValueTask>> _queue;

    public DefaultBackgroundTaskQueue(int capacity)
    {
        BoundedChannelOptions options = new(capacity)
        {
            FullMode = BoundedChannelFullMode.Wait
        };
        _queue = Channel.CreateBounded<Func<CancellationToken, ValueTask>>(options);
    }

    public async ValueTask QueueBackgroundWorkItemAsync(
        Func<CancellationToken, ValueTask> workItem)
    {
        ArgumentNullException.ThrowIfNull(workItem);

        await _queue.Writer.WriteAsync(workItem);
    }

    public async ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(
        CancellationToken cancellationToken)
    {
        Func<CancellationToken, ValueTask>? workItem =
            await _queue.Reader.ReadAsync(cancellationToken);

        return workItem;
    }
}

上記の実装は、キューとして Channel<T> に依存しています。 BoundedChannelOptions(Int32) は、明示的な容量で呼び出されます。容量は、予想されるアプリケーションの負荷と、キューにアクセスする同時実行スレッドの数に基づいて、設定する必要があります。 BoundedChannelFullMode.Wait によって、タスクを返すために ChannelWriter<T>.WriteAsync が呼び出されます。これは、領域が使用可能になった場合にのみ完了します。蓄積を始めるパブリッシャーや呼び出しが多すぎる場合、これによりバックプレッシャが発生します。

Worker クラスを書き換える

次の QueueHostedService の例では以下のようになります。

ProcessTaskQueueAsync メソッドからは、ExecuteAsync で Task が返されます。
ProcessTaskQueueAsync で、キュー内のバックグラウンドタスクがデキューされ、実行されます。
作業項目が待機されてから、StopAsync でサービスが停止します。

既存の Worker クラスを次の C# コードに置き換え、ファイルの名前を QueueHostedService.cs に変更します。

namespace App.QueueService;

public sealed class QueuedHostedService(
        IBackgroundTaskQueue taskQueue,
        ILogger<QueuedHostedService> logger) : BackgroundService
{
    protected override Task ExecuteAsync(CancellationToken stoppingToken)
    {
        logger.LogInformation("""
            {Name} is running.
            Tap W to add a work item to the 
            background queue.
            """,
            nameof(QueuedHostedService));

        return ProcessTaskQueueAsync(stoppingToken);
    }

    private async Task ProcessTaskQueueAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                Func<CancellationToken, ValueTask>? workItem =
                    await taskQueue.DequeueAsync(stoppingToken);

                await workItem(stoppingToken);
            }
            catch (OperationCanceledException)
            {
                // Prevent throwing if stoppingToken was signaled
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "Error occurred executing task work item.");
            }
        }
    }

    public override async Task StopAsync(CancellationToken stoppingToken)
    {
        logger.LogInformation(
            $"{nameof(QueuedHostedService)} is stopping.");

        await base.StopAsync(stoppingToken);
    }
}

MonitorLoop サービスは、w キーが入力デバイスで選択されると常に、ホステッドサービスのためにタスクのエンキューを処理します。

IBackgroundTaskQueue が MonitorLoop サービスに挿入されます。
IBackgroundTaskQueue.QueueBackgroundWorkItemAsync が呼び出され、作業項目がエンキューされます。
作業項目により、実行時間の長いバックグラウンドタスクがシミュレートされます。
- 5 秒間の遅延が 3 回実行されます (Delay)。
- タスクが取り消された場合、try-catch ステートメントによって OperationCanceledException がトラップされます。

namespace App.QueueService;

public sealed class MonitorLoop(
    IBackgroundTaskQueue taskQueue,
    ILogger<MonitorLoop> logger,
    IHostApplicationLifetime applicationLifetime)
{
    private readonly CancellationToken _cancellationToken = applicationLifetime.ApplicationStopping;

    public void StartMonitorLoop()
    {
        logger.LogInformation($"{nameof(MonitorAsync)} loop is starting.");

        // Run a console user input loop in a background thread
        Task.Run(async () => await MonitorAsync());
    }

    private async ValueTask MonitorAsync()
    {
        while (!_cancellationToken.IsCancellationRequested)
        {
            var keyStroke = Console.ReadKey();
            if (keyStroke.Key == ConsoleKey.W)
            {
                // Enqueue a background work item
                await taskQueue.QueueBackgroundWorkItemAsync(BuildWorkItemAsync);
            }
        }
    }

    private async ValueTask BuildWorkItemAsync(CancellationToken token)
    {
        // Simulate three 5-second tasks to complete
        // for each enqueued work item

        int delayLoop = 0;
        var guid = Guid.NewGuid();

        logger.LogInformation("Queued work item {Guid} is starting.", guid);

        while (!token.IsCancellationRequested && delayLoop < 3)
        {
            try
            {
                await Task.Delay(TimeSpan.FromSeconds(5), token);
            }
            catch (OperationCanceledException)
            {
                // Prevent throwing if the Delay is cancelled
            }

            ++ delayLoop;

            logger.LogInformation("Queued work item {Guid} is running. {DelayLoop}/3", guid, delayLoop);
        }

        if (delayLoop is 3)
        {
            logger.LogInformation("Queued Background Task {Guid} is complete.", guid);
        }
        else
        {
            logger.LogInformation("Queued Background Task {Guid} was cancelled.", guid);
        }
    }
}

既存の Program の内容を、次の C# コードに置き換えます。

using App.QueueService;

HostApplicationBuilder builder = Host.CreateApplicationBuilder(args);
builder.Services.AddSingleton<MonitorLoop>();
builder.Services.AddHostedService<QueuedHostedService>();
builder.Services.AddSingleton<IBackgroundTaskQueue>(_ => 
{
    if (!int.TryParse(builder.Configuration["QueueCapacity"], out var queueCapacity))
    {
        queueCapacity = 100;
    }

    return new DefaultBackgroundTaskQueue(queueCapacity);
});

IHost host = builder.Build();

MonitorLoop monitorLoop = host.Services.GetRequiredService<MonitorLoop>()!;
monitorLoop.StartMonitorLoop();

host.Run();

サービスは (Program.cs) に登録されています。ホステッドサービスは、AddHostedService 拡張メソッドを使用して登録されます。 MonitorLoop は、Program.cs のトップレベルステートメントで開始されます。

MonitorLoop monitorLoop = host.Services.GetRequiredService<MonitorLoop>()!;
monitorLoop.StartMonitorLoop();

サービスの登録の詳細については、「.NET での依存関係の挿入」を参照してください。

サービスの機能を確認する

Visual Studio からアプリケーションを実行するには、F5 キーを押すか、[デバッグ]>[デバッグの開始] メニューオプションを選択します。 .NET CLI を使用している場合は、作業ディレクトリから dotnet run コマンドを実行します。

dotnet run

.NET CLI の run コマンドの詳細については、「dotnet run」を参照してください。

プロンプトが表示されたら、 w (または W) を少なくとも 1 回入力して、エミュレートされた作業項目をキューに入れます(例の出力を参照):

info: App.QueueService.MonitorLoop[0]
      MonitorAsync loop is starting.
info: App.QueueService.QueuedHostedService[0]
      QueuedHostedService is running.

      Tap W to add a work item to the background queue.

info: Microsoft.Hosting.Lifetime[0]
      Application started. Press Ctrl+C to shut down.
info: Microsoft.Hosting.Lifetime[0]
      Hosting environment: Development
info: Microsoft.Hosting.Lifetime[0]
      Content root path: .\queue-service
winfo: App.QueueService.MonitorLoop[0]
      Queued work item 8453f845-ea4a-4bcb-b26e-c76c0d89303e is starting.
info: App.QueueService.MonitorLoop[0]
      Queued work item 8453f845-ea4a-4bcb-b26e-c76c0d89303e is running. 1/3
info: App.QueueService.MonitorLoop[0]
      Queued work item 8453f845-ea4a-4bcb-b26e-c76c0d89303e is running. 2/3
info: App.QueueService.MonitorLoop[0]
      Queued work item 8453f845-ea4a-4bcb-b26e-c76c0d89303e is running. 3/3
info: App.QueueService.MonitorLoop[0]
      Queued Background Task 8453f845-ea4a-4bcb-b26e-c76c0d89303e is complete.
info: Microsoft.Hosting.Lifetime[0]
      Application is shutting down...
info: App.QueueService.QueuedHostedService[0]
      QueuedHostedService is stopping.

Visual Studio 内からアプリケーションを実行している場合は、[デバッグ]>[デバッグの停止] を選択します。または、コンソールウィンドウで Ctrl + C キーを押して、キャンセルを通知します。

次の方法で共有

Queue サービスを作成する

前提条件

新しいプロジェクトを作る

キューサービスを作成する

Worker クラスを書き換える

サービスの機能を確認する

関連項目

フィードバック

その他のリソース

次の方法で共有

Queue サービスを作成する

前提条件

新しいプロジェクトを作る

キュー サービスを作成する

Worker クラスを書き換える

サービスの機能を確認する

関連項目

フィードバック

その他のリソース

キューサービスを作成する