如何重新排队失败的后台异步任务

Question

您好，我正在尝试使用BackgroundService 创建一个在后台执行的通用任务队列服务。我避免使用Delegate 函数Func<T1,T2, OuT> 作为EnqueueTask(Task<ResponseHelper> newTask) 方法的输入，因为我想要一个通用的解决方案，所以我选择传递Task<ResponseHelper>。但是这个解决方案不允许我在ExecuteAsync(CancellationToken stoppingToken) 中重新排队失败的任务，因为该任务返回一个ResponseHelper 的实例，它不是失败任务的副本。请帮助我更正代码，我必须从 DequeueTaskAsync(CancellationToken cancellationToken) 函数返回出队任务，而不是 ResponseHelper 的实例。

public interface ITaskQueueHelper
{
    void EnqueueTask(Task<ResponseHelper> newTask);

    Task<ResponseHelper> DequeueTaskAsync(CancellationToken cancellationToken);
}

public class TaskQueueHelper : ITaskQueueHelper
{
    private readonly SemaphoreSlim signal;

    private readonly ConcurrentQueue<Task<ResponseHelper>> taskQueue;

    public TaskQueueHelper()
    {
        signal = new SemaphoreSlim(0);

        taskQueue = new ConcurrentQueue<Task<ResponseHelper>>();
    }

    public void EnqueueTask(Task<ResponseHelper> newTask)
    {
        if (newTask == null)
        {
            throw new ArgumentNullException(nameof(newTask));
        }

        taskQueue.Enqueue(newTask);

        signal.Release();
    }

    public async Task<ResponseHelper> DequeueTaskAsync(CancellationToken cancellationToken)
    {
        await signal.WaitAsync(cancellationToken);

        taskQueue.TryDequeue(out var currentTask);

        /*I need to return currentTask here, instead of an instance of ResponseHelper*/

        return await currentTask;
    }
}

public class TaskQueueService : BackgroundService
{
    private readonly ITaskQueueHelper taskQueue;

    private readonly ILogger<TaskQueueService> logger;

    public TaskQueueService(
        ITaskQueueHelper _taskQueue,
        ILogger<TaskQueueService> _logger)
    {
        logger = _logger;

        taskQueue = _taskQueue;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            ResponseHelper response = await taskQueue.DequeueTaskAsync(stoppingToken);

            try
            {
                if (!response.Status.Equals(ResultCode.Success))
                {
                    /*I need to re-enqueue a failed task here*/

                    //taskQueue.EnqueueTask(currentTask);
                }
            }
            catch (Exception e)
            {
                logger.LogError(e, $"Error occurred executing {nameof(TaskQueueService)}");
            }
        }
    }
}

Answer 1

对于重试，您可以这样做：

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            ResponseHelper response = await taskQueue.DequeueTaskAsync(stoppingToken);

            try
            {
                if (!response.Status.Equals(ResultCode.Success))
                {
                    // Retry the task.
                    response = await taskQueue.DequeueTaskAsync(stoppingToken);

                }
            }
            catch (Exception e)
            {
                logger.LogError(e, $"Error occurred executing {nameof(TaskQueueService)}");
            }
        }
    }