如何正确地从ConcurrentQueue中消费块

本文关键字：ConcurrentQueue 正确地 | 更新日期: 2023-09-27 18:09:44

我需要实现一个可以从多个线程填充的请求队列。当这个队列超过1000个已完成的请求时，应该将这些请求存储到数据库中。下面是我的实现:

public class RequestQueue
{
    private static BlockingCollection<VerificationRequest> _queue = new BlockingCollection<VerificationRequest>();
    private static ConcurrentQueue<VerificationRequest> _storageQueue = new ConcurrentQueue<VerificationRequest>();
    private static volatile bool isLoading = false;
    private static object _lock = new object();
    public static void Launch()
    {
        Task.Factory.StartNew(execute);
    }
    public static void Add(VerificationRequest request)
    {
        _queue.Add(request);
    }
    public static void AddRange(List<VerificationRequest> requests)
    {
        Parallel.ForEach(requests, new ParallelOptions() {MaxDegreeOfParallelism = 3},
            (request) => { _queue.Add(request); });
    }

    private static void execute()
    {
        Parallel.ForEach(_queue.GetConsumingEnumerable(), new ParallelOptions {MaxDegreeOfParallelism = 5}, EnqueueSaveRequest );
    }
    private static void EnqueueSaveRequest(VerificationRequest request)
    {
        _storageQueue.Enqueue( new RequestExecuter().ExecuteVerificationRequest( request ) );
        if (_storageQueue.Count > 1000 && !isLoading)
        {
            lock ( _lock )
            {
                if ( _storageQueue.Count > 1000 && !isLoading )
                {
                    isLoading = true;
                    var requestChunck = new List<VerificationRequest>();
                    VerificationRequest req;
                    for (var i = 0; i < 1000; i++)
                    {
                        if( _storageQueue.TryDequeue(out req))
                            requestChunck.Add(req);
                    }
                    new VerificationRequestRepository().InsertRange(requestChunck);
                    isLoading = false;
                }
            }
        }            
    }
}

有没有办法实现这个没有锁和isLoading?

如何正确地从ConcurrentQueue中消费块

完成您的要求的最简单方法是使用TPL Dataflow库中的块。如

var batchBlock = new BatchBlock<VerificationRequest>(1000);
var exportBlock = new ActionBlock<VerificationRequest[]>(records=>{
               new VerificationRequestRepository().InsertRange(records);
};
batchBlock.LinkTo(exportBlock , new DataflowLinkOptions { PropagateCompletion = true });

就是这样。

你可以用

向起始块发送消息
batchBlock.Post(new VerificationRequest(...));

一旦你完成了你的工作，你可以关闭整个管道，并通过调用batchBlock.Complete();来清除任何剩余的消息，并等待最后一个块完成:

batchBlock.Complete();
await exportBlock.Completion;

BatchBlock将多达1000条记录批量到1000个条目的数组中，并将它们传递给下一个块。ActionBlock默认只使用1个任务，所以它是线程安全的。您可以使用存储库的现有实例，而不必担心跨线程访问:

var repository=new VerificationRequestRepository();
var exportBlock = new ActionBlock<VerificationRequest[]>(records=>{
               repository.InsertRange(records);
};

几乎所有的块都有并发输入缓冲区。每个块在自己的TPL任务上运行，因此每个步骤彼此并发运行。这意味着您可以"免费"获得异步执行，如果您有多个链接的步骤，这可能很重要，例如您使用TransformBlock来修改流经管道的消息。

我使用这样的管道来创建调用外部服务的管道，解析响应，生成最终记录，批处理它们，并通过使用SqlBulkCopy的块将它们发送到数据库。

相关文章：