プログラマ専用SNS ミクプラ

コード:

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

class Program
{
	static void Main ()
	{
		// 標準のキャンセル機構を使うための準備
		var source = new CancellationTokenSource();
		var token = source.Token;

		var currentCount = 0; // 処理中の要素数
		var maxConcurrent = 0; // 処理の最大同時実行数
		
		var workQueue = new BlockingCollection<int>(); // 入力キュー
		var resultQueue = new BlockingCollection<int>(); // 結果キュー

		// ワーカータスク。
		// Producer-Consumer Patternを(BlockingCollectionを通じて)使っているため，それに準じた名前をつけている。
		var consumerTask = Task.Factory.StartNew(() =>
		{
			maxConcurrent =
				workQueue.GetConsumingEnumerable() // 追加されたら取ってくる
				.AsParallel() // 並列化
				.Select(value =>
				{
					// 実際の内部処理

					if (token.IsCancellationRequested)
					{
						// キャンセルされていたら終了する
						return -1;
					}

					var resultValue = Interlocked.Increment(ref currentCount);

					// 重い処理のかわり
					var finish = DateTime.Now + TimeSpan.FromMilliseconds(1000 + value % 1000);
					while (finish > DateTime.Now)
					{
						Thread.SpinWait(10000);
					}

					resultQueue.Add(resultValue);
					Interlocked.Decrement(ref currentCount);

					return resultValue;
				})
				.Max();
			resultQueue.CompleteAdding();

			Console.WriteLine("Consumer Task Finished.");
		});

		// データを生成する
		for (var i = 0; i < 1000; ++i)
		{
			workQueue.Add(i);
		}
		workQueue.CompleteAdding(); // 生成の終了を通知

		Console.ReadLine(); // Enterを押すとキャンセル可能
		source.Cancel();

		Task.WaitAll(consumerTask); // タスクの終了待ち

		Console.WriteLine("Processed : {0}", resultQueue.Count);
		Console.WriteLine("Maximum   : {0}", maxConcurrent);
		if (System.Diagnostics.Debugger.IsAttached)
		{
			Console.ReadLine(); // デバッグ実行時はEnterを待つことで，表示確認を可能にする
		}
	}
}

コード:

Consumer Task Finished.
Processed : 48
Maximum   : 8