Udostępnij za pośrednictwem

Tworzenie usługi kolejki

  • Artykuł

Usługa kolejkowania to doskonały przykład długotrwałej usługi, w której elementy robocze można kolejkować i pracować sekwencyjnie, gdy poprzednie elementy robocze są ukończone. Korzystając z szablonu usługi roboczej, tworzysz nowe funkcje na podstawie elementu BackgroundService.

Z tego samouczka dowiesz się, jak wykonywać następujące czynności:

  • Tworzenie usługi kolejki.
  • Delegowanie pracy do kolejki zadań.
  • Zarejestruj odbiornik klucza konsoli ze zdarzeń IHostApplicationLifetime .

Napiwek

Cały przykładowy kod źródłowy "Pracownicy na platformie .NET" jest dostępny w przeglądarce Samples Browser do pobrania. Aby uzyskać więcej informacji, zobacz Przeglądanie przykładów kodu: Procesy robocze na platformie .NET.

Wymagania wstępne

Tworzenie nowego projektu

Aby utworzyć nowy projekt usługi roboczej za pomocą programu Visual Studio, wybierz pozycję Plik>nowy>projekt....W oknie dialogowym Tworzenie nowego projektu wyszukaj frazę "Usługa procesu roboczego" i wybierz szablon Usługa procesu roboczego. Jeśli wolisz użyć interfejsu wiersza polecenia platformy .NET, otwórz swój ulubiony terminal w katalogu roboczym. dotnet new Uruchom polecenie i zastąp element <Project.Name> odpowiednią nazwą projektu.

dotnet new worker --name <Project.Name>

Aby uzyskać więcej informacji na temat polecenia nowego projektu usługi roboczej interfejsu wiersza polecenia platformy .NET, zobacz dotnet new worker (dotnet new worker).

Napiwek

Jeśli używasz programu Visual Studio Code, możesz uruchomić polecenia interfejsu wiersza polecenia platformy .NET z poziomu zintegrowanego terminalu. Aby uzyskać więcej informacji, zobacz Visual Studio Code: Zintegrowany terminal.

Tworzenie usług kolejkowania

Możesz zapoznać się z QueueBackgroundWorkItem(Func<CancellationToken,Task>) funkcjami z System.Web.Hosting przestrzeni nazw.

Napiwek

Funkcjonalność System.Web przestrzeni nazw celowo nie została przekierowana do platformy .NET i pozostaje wyłączna na platformę .NET Framework. Aby uzyskać więcej informacji, zobacz Wprowadzenie do migracji przyrostowej ASP.NET do ASP.NET Core.

Na platformie .NET, aby modelować usługę inspirowaną QueueBackgroundWorkItem funkcją, zacznij od dodania IBackgroundTaskQueue interfejsu do projektu:

namespace App.QueueService;

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

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

Istnieją dwie metody, które uwidaczniają funkcję kolejkowania, a drugą, która usuwa wcześniej kolejkowane elementy robocze. Element roboczy to Func<CancellationToken, ValueTask>. Następnie dodaj domyślną implementację do projektu.

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;
    }
}

Poprzednia implementacja opiera się na Channel<T> kolejce. Element BoundedChannelOptions(Int32) jest wywoływany z jawną pojemnością. Pojemność powinna być ustawiana na podstawie oczekiwanego obciążenia aplikacji i liczby współbieżnych wątków, które uzyskują dostęp do kolejki. BoundedChannelFullMode.Wait powoduje wywołania zwracania ChannelWriter<T>.WriteAsync zadania, które kończy się tylko wtedy, gdy miejsce stanie się dostępne. Co prowadzi do backpressure, w przypadku zbyt wielu wydawców/wywołań zaczyna gromadzić.

Ponowne zapisywanie klasy Proces roboczy

W poniższym QueueHostedService przykładzie:

  • Metoda ProcessTaskQueueAsync zwraca wartość Task in ExecuteAsync.
  • Zadania w tle w kolejce są usuwane z kolejki i wykonywane w pliku ProcessTaskQueueAsync.
  • Elementy robocze są oczekiwane przed zatrzymanie usługi w programie StopAsync.

Zastąp istniejącą Worker klasę następującym kodem języka C# i zmień nazwę pliku na 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 Usługa obsługuje kolejkowanie zadań dla hostowanej usługi za każdym razem, gdy w klucz jest wybrany na urządzeniu wejściowym:

  • Element IBackgroundTaskQueue jest wstrzykiwany do MonitorLoop usługi.
  • IBackgroundTaskQueue.QueueBackgroundWorkItemAsync element jest wywoływany w celu kolejkowania elementu roboczego.
  • Element roboczy symuluje długotrwałe zadanie w tle: 
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);
        }
    }
}

Zastąp istniejącą Program zawartość następującym kodem języka 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();

Usługi są zarejestrowane w pliku (Program.cs). Hostowana usługa jest zarejestrowana w metodzie AddHostedService rozszerzenia. MonitorLoop jest uruchamiany w instrukcji najwyższego poziomu programu Program.cs :

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

Aby uzyskać więcej informacji na temat rejestrowania usług, zobacz Wstrzykiwanie zależności na platformie .NET.

Weryfikowanie funkcjonalności usługi

Aby uruchomić aplikację z poziomu programu Visual Studio, wybierz pozycję F5 lub wybierz opcję menu Debuguj>rozpocznij debugowanie. Jeśli używasz interfejsu wiersza polecenia platformy .NET, uruchom dotnet run polecenie z katalogu roboczego:

dotnet run

Aby uzyskać więcej informacji na temat polecenia uruchamiania interfejsu wiersza polecenia platformy .NET, zobacz dotnet run.

Po wyświetleniu monitu w wprowadź wartość (lub W) co najmniej raz, aby w kolejce emulowany element roboczy, jak pokazano w przykładowych danych wyjściowych:

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.

W przypadku uruchamiania aplikacji z poziomu programu Visual Studio wybierz pozycję Debuguj zatrzymaj>debugowanie.... Alternatywnie wybierz klawisze Ctrl + C z okna konsoli, aby zasygnalizować anulowanie.

Zobacz też