如何在 c# 中创建一个"fire & forget"异步 FIFO 队列？

Question

我正在尝试异步处理文档。这个想法是用户将文档发送到服务，这需要时间，并且稍后会查看结果(每个文档大约20-90秒(。

理想情况下，我只想填充一些可观察到的集合，这些集合将被系统尽快清空。当有一个项时，处理它并在另一个对象中产生预期的输出，当没有项时，什么都不做。当用户检查输出集合时，他将找到已经处理的项目。

理想情况下，所有项目从一开始都是可见的，并且都有一个状态(已完成、正在进行或在队列中(，但一旦我知道如何做第一个，我就应该能够处理这些状态。

我不确定该用哪个对象，现在我正在看BlockingCollection，但我认为它不适合这份工作，因为当它从另一端清空时，我无法填充它。

private BlockingCollection<IDocument> _jobs = new BlockingCollection<IDocument>();
public ObservableCollection<IExtractedDocument> ExtractedDocuments { get; }
public QueueService()
{
ExtractedDocuments = new ObservableCollection<IExtractedDocument>();
}

public async Task Add(string filePath, List<Extra> extras)
{
if (_jobs.IsAddingCompleted || _jobs.IsCompleted)
_jobs = new BlockingCollection<IDocument>();

var doc = new Document(filePath, extras);
_jobs.Add(doc);
_jobs.CompleteAdding();

await ProcessQueue();
}
private async Task ProcessQueue()
{
foreach (var document in _jobs.GetConsumingEnumerable(CancellationToken.None))
{
var resultDocument = await service.ProcessDocument(document);
ExtractedDocuments.Add(resultDocument );
Debug.WriteLine("Job completed");
}
}

这就是我现在的处理方式。如果我删除CompleteAdding调用，它将在第二次尝试时挂起。如果我有这样的声明，那么我不能只是填满队列，我必须先清空它，这违背了目的。

有没有办法实现我想要实现的目标？一个我将填充的集合，系统将异步和自主处理？

总之，我需要：

• 我可以填充的集合，它将被逐步异步处理。在处理某些文档时，可以添加文档、系列或文档
• 流程完成后将填充的输出集合
• UI线程和应用程序在一切运行时仍能响应
• 我不需要并行处理多个进程，或者一次处理一个文档。无论哪一个最容易放置和维护，都可以(小规模应用(。我想一次一个比较简单

Answer 1

这里的一个常见模式是有一个回调方法，该方法在文档状态更改时执行。在后台任务运行的情况下，它将以最快的速度咀嚼抛出的文档。调用Dispose以关闭处理器。

如果你需要在gui线程上处理回调，你需要将回调同步到你的主线程。如果您正在使用的是Windows窗体，则它具有执行此操作的方法。

这个示例程序实现了所有必要的类和接口，您可以根据需要进行微调和调整。

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
namespace ConsoleApp2
{
class Program
{
private static Task Callback(IExtractedDocument doc, DocumentProcessor.DocState docState)
{
Console.WriteLine("Processing doc {0}, state: {1}", doc, docState);
return Task.CompletedTask;
}
public static void Main()
{
using DocumentProcessor docProcessor = new DocumentProcessor(Callback);
Console.WriteLine("Processor started, press any key to end processing");
for (int i = 0; i < 100; i++)
{
if (Console.KeyAvailable)
{
break;
}
else if (i == 5)
{
// make an error
docProcessor.Add(null);
}
else
{
docProcessor.Add(new Document { Text = "Test text " + Guid.NewGuid().ToString() });
}
Thread.Sleep(500);
}
Console.WriteLine("Doc processor shut down, press ENTER to quit");
Console.ReadLine();
}
public interface IDocument
{
public string Text { get; }
}
public class Document : IDocument
{
public string Text { get; set; }
}
public interface IExtractedDocument : IDocument
{
public IDocument OriginalDocument { get; }
public Exception Error { get; }
}
public class ExtractedDocument : IExtractedDocument
{
public override string ToString()
{
return $"Orig text: {OriginalDocument?.Text}, Extracted Text: {Text}, Error: {Error}";
}
public IDocument OriginalDocument { get; set; }
public string Text { get; set; }
public Exception Error { get; set; }
}
public class DocumentProcessor : IDisposable
{
public enum DocState { Processing, Completed, Error }
private readonly BlockingCollection<IDocument> queue = new BlockingCollection<IDocument>();
private readonly Func<IExtractedDocument, DocState, Task> callback;
private CancellationTokenSource cancelToken = new CancellationTokenSource();
public DocumentProcessor(Func<IExtractedDocument, DocState, Task> callback)
{
this.callback = callback;
Task.Run(() => StartQueueProcessor()).GetAwaiter();
}
public void Dispose()
{
if (!cancelToken.IsCancellationRequested)
{
cancelToken.Cancel();
}
}
public void Add(IDocument doc)
{
if (cancelToken.IsCancellationRequested)
{
throw new InvalidOperationException("Processor is disposed");
}
queue.Add(doc);
}
private void ProcessDocument(IDocument doc)
{
try
{
// do processing
DoCallback(new ExtractedDocument { OriginalDocument = doc }, DocState.Processing);
if (doc is null)
{
throw new ArgumentNullException("Document to process was null");
}
IExtractedDocument successExtractedDocument = DoSomeDocumentProcessing(doc);
DoCallback(successExtractedDocument, DocState.Completed);
}
catch (Exception ex)
{
DoCallback(new ExtractedDocument { OriginalDocument = doc, Error = ex }, DocState.Error);
}
}
private IExtractedDocument DoSomeDocumentProcessing(IDocument originalDocument)
{
return new ExtractedDocument { OriginalDocument = originalDocument, Text = "Extracted: " + originalDocument.Text };
}
private void DoCallback(IExtractedDocument result, DocState docState)
{
if (callback != null)
{
// send callbacks in background
callback(result, docState).GetAwaiter();
}
}
private void StartQueueProcessor()
{
try
{
while (!cancelToken.Token.IsCancellationRequested)
{
if (queue.TryTake(out IDocument doc, 1000, cancelToken.Token))
{
// can chance to Task.Run(() => ProcessDocument(doc)).GetAwaiter() for parallel execution
ProcessDocument(doc);
}
}
}
catch (OperationCanceledException)
{
// ignore, don't need to throw or worry about this
}
while (queue.TryTake(out IDocument doc))
{
DoCallback(new ExtractedDocument { Error = new ObjectDisposedException("Processor was disposed") }, DocState.Error);
}
}
}
}
}