如何在asp.net webapi中只有一个线程用于触发并忘记任务？

Question

我需要在我的 asp.net webapi（框架 .Net 4.7.2）中调用 Redis（使用 StackExchange.Redis），以便以火灾和忘记的方式删除一个键，我正在做一些压力测试。

当我比较各种方式来获得最大速度时：

• 我已经测试了执行带有 FireAndForget 标志的命令，
• 我还通过 await 测量了一个对 Redis 的简单命令。
• 我现在正在寻找一种方法来收集在 15 毫秒的窗口中收到的命令列表，并通过管道将它们全部执行。

我首先尝试使用 Task.Run Action 来调用 Redis，但我观察到的问题是在压力下，我的 webapi 的内存不断攀升。 内存中充满了 System.Threading.IThreadPoolWorkItem[] 对象，代码如下：

    [HttpPost]
    [Route("api/values/testpostfireforget")]
    public  ApiResult<int> DeleteFromBasketId([FromBody] int basketId)
    {

        var response = new DeleteFromBasketResponse<int>();

        var cpt = Interlocked.Increment(ref counter);

            Task.Run(async () => {
                await db.StringSetAsync($"BASKET_TO_DELETE_{cpt}",cpt.ToString())
                         .ConfigureAwait(false); 
            });  
           return response;
     }

所以我认为在压力下，我的 api 会继续在内存中排队后台任务，并尽可能快地一个接一个地执行它们，但比传入的请求要少...

所以我正在寻找一种方法让 只有一个 长期运行的后台线程与 asp.net webapi 一起运行，它可以捕获发送到 Redis 的命令并通过管道执行它们他们。

我正在考虑通过实现 IHostedService 接口来运行后台任务，但在这种情况下，后台任务似乎不会与我当前的 http 请求共享任何状态。因此，实现 IhostedService 对于计划的后台任务会很方便，但在我的情况下不是这样，或者我不知道如何......

Answer 1

基于StackExchange.Redis documentation，您可以使用CommandFlags.FireAndForget 标志：

[HttpPost]
[Route("api/values/testpostfireforget")]
public  ApiResult<int> DeleteFromBasketId([FromBody] int basketId)
{

    var response = new DeleteFromBasketResponse<int>();
    var cpt = Interlocked.Increment(ref counter);

    db.StringSet($"BASKET_TO_DELETE_{cpt}", cpt.ToString(), flags: CommandFlags.FireAndForget);  

    return response;
}

编辑 1：基于评论的另一种解决方案

您可以使用发布/订阅方法。像这样的东西应该可以工作：

public class MessageBatcher
{
    private readonly IDatabase target;
    private readonly BlockingCollection<Action<IDatabaseAsync>> tasks = new();
    private Task worker;

    public MessageBatcher(IDatabase target) => this.target = target;

    public void AddMessage(Action<IDatabaseAsync> task) => tasks.Add(task);
    
    public IDisposable Start(int batchSize)
    {
        var cancellationTokenSource = new CancellationTokenSource();
        worker = Task.Factory.StartNew(state =>
        {
            var count = 0;
            var tokenSource = (CancellationTokenSource) state;
            var box = new StrongBox<IBatch>(target.CreateBatch());
            tokenSource.Token.Register(b => ((StrongBox<IBatch>)b).Value.Execute(), box);
            foreach (var task in tasks.GetConsumingEnumerable(tokenSource.Token))
            {
                var batch = box.Value;
                task(batch);
                if (++count == batchSize)
                {
                    batch.Execute();
                    box.Value = target.CreateBatch();
                    count = 0;
                }
            }
            
        }, cancellationTokenSource, cancellationTokenSource.Token, TaskCreationOptions.LongRunning, TaskScheduler.Current);

        return new Disposer(worker, cancellationTokenSource);
    }

    private class Disposer : IDisposable
    {
        private readonly Task worker;
        private readonly CancellationTokenSource tokenSource;

        public Disposer(Task worker, CancellationTokenSource tokenSource) => (this.worker, this.tokenSource) = (worker, tokenSource);

        public void Dispose()
        {
            tokenSource.Cancel();
            worker.Wait();
            tokenSource.Dispose();
        }
    }
}

用法：

private readonly MessageBatcher batcher;

ctor(MessageBatcher  batcher) // ensure that passed `handler` is singleton and already already started
{
    this.batcher= batcher;
}

[HttpPost]
[Route("api/values/testpostfireforget")]
public  ApiResult<int> DeleteFromBasketId([FromBody] int basketId)
{
    var response = new DeleteFromBasketResponse<int>();
    var cpt = Interlocked.Increment(ref counter);

    batcher.AddMessage(db => db.StringSetAsync($"BASKET_TO_DELETE_{cpt}", cpt.ToString(), flags: CommandFlags.FireAndForget));  

    return response;
}