如何设计支持泛型Func<Task<T>>的限频API请求队列？

问题背景 #

我需要实现一个支持限频外部API调用的队列系统，要求将不同返回类型的API请求放入队列逐个执行，但当前设计中，队列处理器无法确定队列项的具体类型来执行对应的函数，且不想使用object或反射来解决。

当前的队列和队列项设计如下：

public class MyQueue // 单例
{
    readonly ConcurrentQueue<MyQueueItem> queue = [];

    public async Task<T> EnqueueAndWaitAsync<T>(Func<Task<T>> func)
    {
        var item = new MyQueueItem<T>(func);
        queue.Enqueue(item);
        return await item.Task;
    }

    public async Task<MyQueueItem> DequeueAsync()
    {
        while (true)
        {
            if (queue.TryDequeue(out var item))
            {
                return item;
            }
            await Task.Delay(100);
        }
    }
}

public class MyQueueItem { }

public class MyQueueItem<T>(Func<Task<T>> func) : MyQueueItem
{
    public Func<Task<T>> Func => func;

    readonly TaskCompletionSource<T> tcs = new();
    public Task<T> Task => tcs.Task;
}

API服务的调用方式：

public class MyApi(MyQueue queue)
{
    public async Task<int> GetInt() => await queue.EnqueueAndWaitAsync(async () =>
    {
        await Task.Delay(1000);
        return 1;
    });

    public async Task<string> GetString() => await queue.EnqueueAndWaitAsync(async () =>
    {
        await Task.Delay(2000);
        return "Hello";
    });
}

核心问题：队列处理器MyQueueProcessor无法确定出队项的具体类型，无法调用对应的Func()执行逻辑：

public class MyQueueProcessor(MyQueue queue)
{
    public async Task DoWorkAsync()
    {
        while (true) // 生产环境会在此处理限频等待
        {
            var item = await queue.DequeueAsync();
            // 如何确定类型并调用Func()？
        }
    }
}

解决方案：利用多态封装执行逻辑 #

通过给基类MyQueueItem添加抽象执行方法，让泛型子类自行实现具体的执行和结果回调逻辑，处理器只需调用统一的方法即可，完全无需反射或类型转换。

修改后的代码 #

1. 修改队列项基类与泛型子类

// 改为抽象类，添加统一的执行方法
public abstract class MyQueueItem
{
    public abstract Task ExecuteAsync();
}

public class MyQueueItem<T>(Func<Task<T>> func) : MyQueueItem
{
    private readonly Func<Task<T>> _func = func;
    private readonly TaskCompletionSource<T> _tcs = new();

    public Task<T> Task => _tcs.Task;

    // 实现基类的执行方法，自行处理逻辑和结果传递
    public override async Task ExecuteAsync()
    {
        try
        {
            var result = await _func();
            _tcs.SetResult(result);
        }
        catch (Exception ex)
        {
            // 捕获异常并传递给等待的调用方
            _tcs.SetException(ex);
        }
    }
}

2. 修改队列处理器

public class MyQueueProcessor(MyQueue queue)
{
    public async Task DoWorkAsync()
    {
        while (true)
        {
            var item = await queue.DequeueAsync();
            // 直接调用统一的ExecuteAsync，无需关心具体类型
            await item.ExecuteAsync();
        }
    }
}

方案优势 #

• 完全基于多态实现，类型安全，避免了反射或object转换带来的性能问题和类型风险
• 每个队列项自行负责执行逻辑和结果/异常的传递，职责清晰
• 处理器代码简洁，无需关心具体的返回类型，专注于队列调度和限频逻辑

内容的提问来源于stack exchange，提问作者Luke Vo