正确使用ConcurrentQueue中的块

我需要实现一个可以从多个线程填充的请求队列.当此队列变得大于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的实现方法？

解决方法:

执行要求的最简单方法是使用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的块的数据库中.

标签：producer-consumer,multithreading,task-parallel-library,tpl-dataflow,c
来源： https://codeday.me/bug/20191026/1938062.html