Поделиться через

Создание службы очередей

Служба очередей — это отличный пример длительно работающей службы, где задания могут быть помещены в очередь и обрабатываться последовательно по мере завершения предыдущих заданий. Опираясь на шаблон рабочей службы BackgroundService, вы создаете новые функциональные возможности.

В этом руководстве вы узнаете, как:

  • Создайте службу очередей.
  • Делегировать работу в очередь задач.
  • Регистрация прослушивателя ключей консоли из IHostApplicationLifetime событий.

Подсказка

Все исходные коды примеров "Workers in .NET" доступны в Samples Browser для скачивания. Дополнительные сведения см. в разделе Примеры кода: рабочие процессы в .NET.

Предпосылки

Создание нового проекта

Чтобы создать проект рабочей службы с помощью Visual Studio, выберите файл>нового>проекта.... В диалоговом окне "Создание проекта " найдите "Рабочая служба" и выберите шаблон "Рабочая служба". Если вы предпочитаете использовать интерфейс командной строки .NET, откройте любимый терминал в рабочем каталоге. Выполните команду dotnet new, и замените <Project.Name> на желаемое имя проекта.

dotnet new worker --name <Project.Name>

Для получения дополнительной информации о команде создания нового проекта рабочей службы в .NET CLI, см. dotnet new worker.

Подсказка

Если вы используете Visual Studio Code, вы можете запускать команды CLI .NET из интегрированного терминала. Дополнительные сведения см. в visual Studio Code: интегрированный терминал.

Создание служб очередей

Вы можете быть знакомы с QueueBackgroundWorkItem(Func<CancellationToken,Task>) функциональностью из System.Web.Hosting пространства имен.

Подсказка

Функциональные возможности System.Web пространства имен намеренно не переносились на .NET и остаются эксклюзивными для .NET Framework. Дополнительные сведения см. в статье "Начало работы с поэтапной миграцией на 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);
}

Существует два метода: один предоставляет функции очереди, а другой извлекает из очереди ранее помещенные в нее рабочие элементы. Рабочий элемент — это 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 возвращает значение Task в ExecuteAsync.
  • Фоновые задачи в очереди выводятся из очереди и выполняются в 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-секундные задержки 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, выполните dotnet run команду из рабочего каталога:

dotnet run

Дополнительные сведения о команде выполнения .NET CLI см. в разделе dotnet run.

При появлении запроса введите w (или W) по крайней мере один раз для постановки в очередь эмулируемого задания, как показано в примере выходных данных.

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 в окне консоли, чтобы сообщить об отмене.

См. также

  • Last updated on