连续作业队列在关闭之前完成作业

问题描述

使用以下代码，我使用具有连续作业队列的.net核心后台服务，其中ExecuteAsync内部模拟了已添加到队列中的作业（这可能是收集订单，产生订单响应等）>

后台服务：

public class Worker : BackgroundService
{

    public WorkerJobQueue orderQueue { get; set; }

    public override async Task StartAsync(CancellationToken cancellationToken)
    {
        Console.WriteLine("Sales_Order_Processor_Service Starting");
        
        orderQueue = new WorkerJobQueue();
        orderQueue.RegisterHandler<TestJob>(TestJobWorker.DoJob);

        await base.StartAsync(cancellationToken);
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        Console.WriteLine("Sales_Order_Processor_Service Stopping");
        await orderQueue.EndQueue(cancellationToken);
        await base.StopAsync(cancellationToken);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var i = 0;

        while (!stoppingToken.IsCancellationRequested)
        {
            Console.ReadLine();

            for (var j = 0; j < 50; j++)
            {
                var tmp = new TestJob { JobNumber = i };
                Console.WriteLine($"Adding job {tmp.JobNumber} to queue");
                await orderQueue.Enqueue(tmp);
                i++;
            }

            Console.WriteLine($"{orderQueue.GetNumberOfRemainingJobs()} Jobs in queue...");

        }
    }

}

工人工作队列：

public class WorkerJobQueue
{

    private ActionBlock<IJob> _workerBlock;

    public WorkerJobQueue()
    {
 
    }

    public void RegisterHandler<T>(Action<T> handleAction) where T : IJob
    {
        Action<IJob> actionWrapper = (job) => handleAction((T)job);
        var executionDataflowBlockOptions = new ExecutionDataflowBlockOptions()
        {
            MaxDegreeOfParallelism = 5,};
        _workerBlock = new ActionBlock<IJob>((job) => actionWrapper(job),executionDataflowBlockOptions);
    }

    public async Task Enqueue(IJob job)
    {
        await _workerBlock.SendAsync(job);
    }

    public int GetNumberOfRemainingJobs()
    {
        return _workerBlock.InputCount;
    }

    public async Task EndQueue(CancellationToken stoppingToken)
    {
        await Task.WhenAll(_workerBlock.Completion,Task.Delay(Timeout.Infinite,stoppingToken));
    }
}

工作项：

    public class TestJob : IJob
{
    public int JobNumber { get; set; }

}

并且只是为了模拟工作完成的一些工作：

    public class TestJobWorker
{
   
    public static void DoJob(TestJob testJob)
    {
        var rnd = new Random();
        var ranNum = rnd.Next(10);

        Console.WriteLine($"Starting job {testJob.JobNumber} sleeping for {ranNum} seconds");
        System.Threading.Thread.Sleep(ranNum* 1000);
        Console.WriteLine($"Finished job {testJob.JobNumber}");
    }

}

作业队列可以正常工作，因为它应该在按键时将50个作业添加到队列中，但是每当服务停止/控制台窗口关闭时，它实际上并没有等待作业队列完成，也没有进入StopAsync功能？

我想念过在关闭/停止服务时调用函数StopAsync吗？还是队列中的逻辑实际上在结束队列时不正确？

解决方法

结果是，StopAsync的调用比托管stop的调用晚，我使用以下代码触发停止：

        private readonly IHostApplicationLifetime _hostApplicationLifetime;
        public Worker(IHostApplicationLifetime hostApplicationLifetime)
        {
            _hostApplicationLifetime = hostApplicationLifetime;
        }

        private void OnStopping()
        {
            orderQueue.EndQueue();
            logger.Debug("Sales_Order_Processor_Service Stopping");
        }

.net-core background-service tpl-dataflow