asp.net-core - 具有队列处理功能的 .NET 核心 Web api

Question

如何设置 .NET 核心 Web api

接受字符串值，

放入队列

并返回消息被接受的标志(不管它是否被处理)。

此外，还有一个例程不断检查队列，并一一处理消息。

根据要求，api将充当消息的接收者，每分钟可能会被点击数百次，而它接收到的消息应该一个一个地处理。
我对 web apis 有点陌生，所以想知道这样的设置是否很好，如果可以，如何将不同的组件放在一起。

提前致谢..

Answer 1

老实说，我认为在一个进程中接收和处理消息没有意义，所以我建议使用外部消息系统，如 RabbitMQ或 Kafka或您喜欢的任何其他现有系统，您可以在其中放置消息，而另一个进程将使用它。话题挺大的，可以从this tutorial开始

如果您仍然想在一个进程中使用它也是可能的，您可以创建一个后台任务队列，将您的消息放在那里并创建 background task这将从该队列中消耗它们。

public interface IBackgroundTaskQueue
{
    void QueueBackgroundWorkItem(Func<CancellationToken, Task> workItem);

    Task<Func<CancellationToken, Task>> DequeueAsync(
        CancellationToken cancellationToken);
}

public class BackgroundTaskQueue : IBackgroundTaskQueue
{
    private ConcurrentQueue<Func<CancellationToken, Task>> _workItems = 
        new ConcurrentQueue<Func<CancellationToken, Task>>();
    private SemaphoreSlim _signal = new SemaphoreSlim(0);

    public void QueueBackgroundWorkItem(
        Func<CancellationToken, Task> workItem)
    {
        if (workItem == null)
        {
            throw new ArgumentNullException(nameof(workItem));
        }

        _workItems.Enqueue(workItem);
        _signal.Release();
    }

    public async Task<Func<CancellationToken, Task>> DequeueAsync(
        CancellationToken cancellationToken)
    {
        await _signal.WaitAsync(cancellationToken);
        _workItems.TryDequeue(out var workItem);

        return workItem;
    }
}

后台任务:

public class QueuedHostedService : BackgroundService
{
    private readonly ILogger _logger;

    public QueuedHostedService(IBackgroundTaskQueue taskQueue, 
        ILoggerFactory loggerFactory)
    {
        TaskQueue = taskQueue;
        _logger = loggerFactory.CreateLogger<QueuedHostedService>();
    }

    public IBackgroundTaskQueue TaskQueue { get; }

    protected async override Task ExecuteAsync(
        CancellationToken cancellationToken)
    {
        _logger.LogInformation("Queued Hosted Service is starting.");

        while (!cancellationToken.IsCancellationRequested)
        {
            var workItem = await TaskQueue.DequeueAsync(cancellationToken);

            try
            {
                await workItem(cancellationToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, 
                   $"Error occurred executing {nameof(workItem)}.");
            }
        }

        _logger.LogInformation("Queued Hosted Service is stopping.");
    }
}

登记:

public void ConfigureServices(IServiceCollection services)
{
    services.AddHostedService<QueuedHostedService>();
    services.AddSingleton<IBackgroundTaskQueue, BackgroundTaskQueue>();
}

注入(inject) Controller :

public class ApiController
{
    private IBackgroundTaskQueue queue;
    public ApiController(IBackgroundTaskQueue queue)
    {
        this.queue = queue;
    }

    public IActionResult StartProcessing()
    {
        queue.QueueBackgroundWorkItem(async token =>
        {
            // put processing code here
        }

        return Ok();
    }
}

您可以修改 BackgroundTaskQueue 以满足您的要求，但我希望您了解这背后的想法。