.NET5 后台服务并发

Question

我有一个后台服务，它将在应用程序启动时启动。后台服务将根据设置的工人数量开始创建多个任务。当我进行各种试验并监视 DB 上的打开连接时。打开的连接总是与我设置的工人相同。假设我设置了 32 个工作人员，那么连接将始终显示为 32 个打开的连接，因为我使用查询来检查它。仅供参考，我使用 Postgres 作为数据库服务器。为了检查打开的连接，我使用下面的查询来检查应用程序运行时的连接。

select * from pg_stat_activity where application_name = 'myapplication';

下面是后台服务代码。

public class MessagingService : BackgroundService {
     private int worker = 32;

     protected override async Task ExecuteAsync(CancellationToken cancellationToken) {
         var tasks = new List<Task>();
         for (int i=0; i<worker; i++) {
         tasks.Add(DoJob(cancellationToken));
         }
         while (!cancellationToken.IsCancellationRequested) {
            try {
               var completed = await Task.WhenAny(tasks);
               tasks.Remove(completed);
            } catch (Exception) {
               await Task.Delay(1000, cancellationToken);
            }
            if (!cancellationToken.IsCancellationRequested) {
               tasks.Add(DoJob(cancellationToken));
            }
        }
     }
    private async Task DoJob(CancellationToken cancellationToken) {
        using (var scope = _services.CreateScope()) {
            var service = scope.ServiceProvider
                .GetRequiredService<MessageService>();
            try {
                //do select and update query on db if null return false otherwise send mail
                if (!await service.Run(cancellationToken)) {
                    await Task.Delay(1000, cancellationToken);
                }
            } catch (Exception) {
                await Task.Delay(1000, cancellationToken);
            }
        }
    }
}   

工作流程不正确，因为它将继续创建任务并使连接保持打开和空闲状态。此外，运行这些任务时 CPU 和内存使用率很高。当在 DB 上找不到任何记录时，我如何才能实现目前仅保持 1 个工作人员运行？如果找到一条或更多记录，它将继续增加，直到预设的最大工作人员，然后当记录小于最大工作人员时减少工作人员。如果这个问题太模糊或基于意见，请告诉我，我会尽力使其尽可能具体。

更新目的

此服务的目的是执行电子邮件传递。还有另一个 API 将用于创建计划作业。将作业添加到数据库后，此服务将在预定时间发送电子邮件。例如，将 5k 计划作业添加到数据库中，计划执行作业的时间为 '2021-12-31 08:00:00'，创建计划作业的时间为 2021-12-31 00:00:00 '。该服务将从 00:00:00 一直循环到 08:00:00，同时运行 32 名工作人员，然后开始进行电子邮件传递。当没有安排只有 1 个工作人员正在运行的作业时，我怎样才能像往常一样提高它的效率。当它检查有 5k 计划作业时，它将充分利用所有工作人员。完成 5k 个作业后，将返回 1 个工人。

Answer 1

我的建议是通过使用来自TPL Dataflow 库的ActionBlock<T>，让自己免于手动创建和维护工作任务的负担。该组件是输入队列和Action<T> 委托的组合。您在其构造函数中指定委托，并使用其Post 方法向其提供消息。组件以指定的并行度为其接收到的每条消息调用委托。当没有更多消息要发送时，您通过调用其Complete 方法通知它，然后调用await 其Completion，这样您就知道委派给它的所有工作都已完成。

下面是一个粗略的演示，如果你可以使用这个组件：

protected override async Task ExecuteAsync(CancellationToken cancellationToken)
{
    var processor = new ActionBlock<Job>(async job =>
    {
        await ProcessJob(job);
        await MarkJobAsCompleted(job);
    }, new ExecutionDataflowBlockOptions()
    {
        MaxDegreeOfParallelism = 32
    });

    try
    {
        while (true)
        {
            Task delayTask = Task.Delay(TimeSpan.FromSeconds(60), cancellationToken);
            Job[] jobs = await FetchReadyToProcessJobs();
            foreach (var job in jobs)
            {
                await MarkJobAsPending(job);
                processor.Post(job);
            }
            await delayTask; // Will throw when the token is canceled
        }
    }
    finally
    {
        processor.Complete();
        await processor.Completion;
    }
}

FetchReadyToProcessJobs 方法应该连接到数据库，并获取所有需要处理的作业。在上面的示例中，此方法每 60 秒调用一次。 Task.Delay 在调用该方法之前创建，并在返回的作业发布到ActionBlock<T> 之后等待。这样调用之间的间隔将是稳定和一致的。