BlockingCollection の使い方


コードリーディングしている時に`BlockingCollection` というものが出てきてよくわからなかったので、調べてみました。主に[BlockingCollection in C# - Introduction and Examples](http://dotnetpattern.com/csharp-blockingcollection#:~:text=%20BlockingCollection%20in%20C%23%20-%20Introduction%20and%20Examples,new%20items.%20Add%20method%20takes%20a...%20More%20)を参考にさせていただいて、自分のコードを書いてみました。

# Blocking Collection とは
`BlockingCollection` はProducer-Consumer パターンの実装です。`IPublisherConsumerCollection<T>` というインターフェイスがありますが、そのスレッドセーフの実装です。つまり、コンカレントな状況においてもちゃん動いてくれるものです。

デフォルトでは内部的に`ConcurrentQueue` を用いているようですが、コンストラクタで`IProducerConsumerCollection<T>` を実装したクラスを渡すことで、そちらのクラスを使用するように変えることもできるようです。インターフェイスは次のものになっています。

_IProducerConsumerCollection<T>_

```csharp
public interface IProducerConsumerCollection<T> : IEnumerable<T>, IEnumerable, ICollection
{
    void CopyTo(T[] array, int index);
    T[] ToArray();
    bool TryAdd(T item);
    bool TryTake([MaybeNullWhen(false)] out T item);
}
```

## コンストラクタ

実際に挙動を確認してみましょう。`BlockingCollection` クラスを単純にインスタンス生成します。`boundedCapacity` を渡すことで、このコレクションの最大値を指定することができます。先ほど述べた通り、ここで、`ConcurrentQueue` 以外の実装を渡すことも可能です。

```csharp
var blockingCollection = new BlockingCollection<string>(boundedCapacity: 3);
```

## Producer  

コンカレントOKとの事なので、スレッドを生成して、そこで、`Add` を使って要素を足していきます。ここでは、コンソールから入力したものを渡しています。挙動としてのポイントは、上記のコンストラクタで指定した、`boundedCapacity` を超えると、`Add` メソッドがブロックして、Consumer がアイテムを取得してくれるのを待ちます。

```csharp
Task producerThread = Task.Factory.StartNew(() =>
{
    while (true)
    {
        string command = Console.ReadLine();
        if (command.Contains("quit")) break;
        blockingCollection.Add(command);   // blocked if it reach the capacity
    }
});
```

この待ち受けの挙動が好きではない場合、`TryAdd` メソッドもあります。このメソッドの場合、一定の時間ブロックされたら、「失敗した」とみなして処理をさせることも可能です。`CancellationToken` を持つオーバーロードもあります。

```csharp
if (blockingCollection.TryAdd(command, TimeSpan.FromSeconds(1)))
{
        // it works!
}
else
{
    Console.WriteLine($"It reached boundedCapacity: {capacity} couldn't add {command}");
}
```

## Consumer

`Take` メソッドにより、１件のアイテムを取得することができます。もし、`BlockingCollection` のインスタンスに１件もなかったら、ここでブロックされます。`blockingCollection.IsComplete` メソッドで、`BlockingCollection` が終了したことの通知を受け取ることができます。


### NOTE
ちなみにこのサンプルで、`.GetAwaiter().GetResult()` みたいなクソださなことをしているかというと、`Task.Factory.StartNew(async () => {}` にすると、async のため labmda　の実行がブロックされて、すぐに終了したとみなされて、後に出てくる `WaitAll` メソッドでこのスレッドの終了を待ち受ける処理がうまく動作しなくなるからです。正直もっといいやり方がありそう。本番では、async/await を使うので直接スレッドを起動していないので、問題にはなっていませんが、、ダサさを何とかしたい。

```csharp
Task consumerAThread = Task.Factory.StartNew(() =>
{
    while (true)
    {
        if (blockingCollection.IsCompleted) break;
        string command = blockingCollection.Take();
        Console.WriteLine($"ConsumerA: Take Received: {command}");
        Task.Delay(TimeSpan.FromSeconds(10)).GetAwaiter().GetResult();
    }
});
```

`TryTake` メソッドは、ブロック中に、一定の時間がたつと失敗させてくれるメソッドです。

```csharp
Task consumerBThread = Task.Factory.StartNew(() =>
{
    while (true)
    {
        if (blockingCollection.IsCompleted) break;
        string command;
        if (blockingCollection.TryTake(out command, TimeSpan.FromSeconds(5)))
        {
            Console.WriteLine($"ConsumerB: TryTake Received: {command}");
        }
        else
        {
            Console.WriteLine($"consumerB: Can't take now.");
        }
        Task.Delay(TimeSpan.FromSeconds(10)).GetAwaiter().GetResult();
    }
});
```

## CancellationToken

`TryXXX` メソッドは、`CancellationToken`をサポートしているので、活用することもできます。`CancellationToken` が発行されたら、`OperationCanceledException` がスローされます。


```csharp
CancellationTokenSource source = new CancellationTokenSource();
```

```csharp
Task consumerBThread = Task.Factory.StartNew(() =>
{
    while (true)
    {
        if (blockingCollection.IsCompleted) break;
        string command;
        try
        {
            if (blockingCollection.TryTake(out command, (int)TimeSpan.FromSeconds(5).TotalMilliseconds, source.Token))
            {
                Console.WriteLine($"ConsumerB: TryTake Received: {command}");
            }
            else
            {
                Console.WriteLine($"consumerB: Can't take now.");
            }
        } catch (OperationCanceledException e)
        {
            Console.WriteLine($"ConsumerB: Task is cancelled.: {e.Message}");
            break;
        }
        Task.Delay(TimeSpan.FromSeconds(10)).GetAwaiter().GetResult();
    }
}); 

```

プログラムの全体像をシェアしておきます。実行して、何かを入力すると、Producerが`BlockingCollection` にアイテムを代入していきます。`Producer` , `ConsumerA`, `ConsumerB` のブロッキングの振る舞いを観察することができます。`cancel` とタイプすると、`CancellationToken` が発行されて終了します。もしくは、`quit` で終了します。

```csharp
class Program
{
    static void Main(string[] args)
    {
        int capacity = 3;
        // Blocking Collection 
        var blockingCollection = new BlockingCollection<string>(boundedCapacity: capacity);
        CancellationTokenSource source = new CancellationTokenSource();

        Task producerThread = Task.Factory.StartNew(() =>
        {
            while (true)
            {
                string command = Console.ReadLine();
                if (command.Contains("quit")) break;
                if (command.Contains("cancel"))
                {
                    Console.WriteLine("Cancelling ...");
                    source.Cancel();
                    break;
                }
                // blockingCollection.Add(command);   // blocked if it reach the capacity
                if (blockingCollection.TryAdd(command, TimeSpan.FromSeconds(1)))
                {
                        // it works!
                }
                else
                {
                    Console.WriteLine($"It reached boundedCapacity: {capacity} couldn't add {command}");
                }
            }
        });
        Task consumerAThread = Task.Factory.StartNew(() =>
        {
            while (true)
            {
                if (blockingCollection.IsCompleted) break;
                string command = blockingCollection.Take();
                Console.WriteLine($"ConsumerA: Take Received: {command}");
                Task.Delay(TimeSpan.FromSeconds(10)).GetAwaiter().GetResult();
            }
        });
        Task consumerBThread = Task.Factory.StartNew(() =>
        {
            while (true)
            {
                if (blockingCollection.IsCompleted) break;
                string command;
                try
                {
                    if (blockingCollection.TryTake(out command, (int)TimeSpan.FromSeconds(5).TotalMilliseconds, source.Token))
                    {
                        Console.WriteLine($"ConsumerB: TryTake Received: {command}");
                    }
                    else
                    {
                        Console.WriteLine($"consumerB: Can't take now.");
                    }
                } catch (OperationCanceledException e)
                {
                    Console.WriteLine($"ConsumerB: Task is cancelled.: {e.Message}");
                    break;
                }
                Task.Delay(TimeSpan.FromSeconds(10)).GetAwaiter().GetResult();
            }
        });
        Task.WaitAll(producerThread, consumerAThread, consumerBThread);
    }
}
```

# Source
今回のサンプルです。
* [Sample](https://github.com/TsuyoshiUshio/BlockingCollectionSpike)