Making PLINQ and BlockingCollection work together
I have put together a simple application that monitors file creation events, creates some objects from the files content, and does some processing. Here is the sample code:
class Program
{
private const string Folder = "C:\\Temp\\InputData";
static void Main(string[] args)
{
var cts = new CancellationTokenSource();
foreach (var obj in Input(cts.Token))
Console.WriteLine(obj);
}
public static IEnumerable<object> Input(CancellationToken cancellationToken)
{
var fileList = new BlockingCollection<string>();
var watcher = new FileSystemWatcher(Folder);
watcher.Created += (source, e) =>
{
if (cancellationToken.IsCancellationRequested)
watcher.EnableRaisingEvents = false;
else if (Path.GetFileName(e.FullPath) == "STOP")
{
watcher.EnableRaisingEvents = false;
fileList.CompleteAdding();
File.Delete(e.FullPath);
}
else
fileList.Add(e.FullPath);
};
watcher.EnableRaisingEvents = true;
return from file in
fileList.GetConsumingEnumerable(cancellationToken)
//.AsParallel()
//.WithCancellation(cancellationToken)
//.WithDegreeOfParallelism(5)
let obj = CreateM开发者_JAVA百科yObject(file)
select obj;
}
private static object CreateMyObject(string file)
{
return file;
}
}
It all works fine, but when I uncomment AsParallel (and the next two lines) it doesn't yield results right away. This delay is probably caused by PLINQ partitioning? However, I expect this query to yield items as soon as they are added to the BlockingCollection. Is this possible to achieve using PLINQ?
That is what .WithMergeOptions(ParallelMergeOptions.NotBuffered)
should be designed for.
精彩评论