マルチスレッドでキューやスタックなどを利用するには？［.NET 4.0以降、C#／VB］：.NET TIPS

Imports System.Collections.Concurrent
Imports System.Console

Const MAX_QUE As Integer = 3 ' コレクションに保持する最大個数

' キュー（先入れ先出し）を作る
Dim que = New BlockingCollection(Of Integer)(MAX_QUE)
' または、
Dim que = New BlockingCollection(Of Integer)(
                New ConcurrentQueue(Of Integer)(), MAX_QUE)

' スタック（後入れ先出し）を作る
Dim stack = New BlockingCollection(Of Integer)(
                  New ConcurrentStack(Of Integer)(), MAX_QUE)

' データバッグ（取り出し順序不定）を作る
Dim bag = New BlockingCollection(Of Integer)(
                New ConcurrentBag(Of Integer)(), MAX_QUE)

Dim data As Integer = 1

' キューへデータを投入する（ロック不要）
Dim addSuccess As Boolean = que.TryAdd(data, Threading.Timeout.Infinite)

' キューからデータを取り出す（ロック不要）
Dim takeSuccess As Boolean = que.TryTake(data, Threading.Timeout.Infinite)

For Each dat In que.GetConsumingEnumerable()
  ' 取り出したデータを使う
Next

Const MAX_QUE As Integer = 3 ' キューに保持する最大個数

' キューを生成する
Dim que = New BlockingCollection(Of Integer)(MAX_QUE)

Try ' BlockingCollectionはDisposeしなければならないので、try〜finallyする

  ' Producerを定義して実行開始
  Dim produceTask = Task.Run(
  Async Function()
    ' 1から5まで順にデータを投入していく
    For i As Integer = 1 To 5
      ' データ投入が間に合わない場合をシミュレート
      If (i = 5) Then
        WriteLine($"stop enqueue")
        Await Task.Delay(1000)
        WriteLine($"restart enqueue")
      End If

      ' 投入データを作成する処理をシミュレート
      Await Task.Delay(10)

      ' キューへデータを投入する（ロック不要）
      que.TryAdd(i, Threading.Timeout.Infinite)
      ' ↑Infinite（無限）指定は、キューが満杯のときは空くまでブロックされる
      WriteLine($"Enqueue({i}), count={que.Count}")
    Next
    que.CompleteAdding() ' データ投入完了を通知
    WriteLine($"end enqueue")
  End Function)

  ' Consumerを定義（その1−TryTakeメソッドを使って取り出し）
  Dim consume1 As Func(Of String, Task) =
  Async Function(header)
    While (Not que.IsCompleted)
      Dim data As Integer
      ' キューからデータを取り出す（ロック不要）
      Dim success As Boolean = que.TryTake(data, Threading.Timeout.Infinite)
      ' ↑Infinite（無限）指定は、キューが空のとき、
      ' データが投入されるかque.IsCompletedがtrueになるまでブロックされる
      If (success) Then
        WriteLine($"{header} Dequeue({data}), count={que.Count}")

        ' キューから取り出したデータを使う処理をシミュレート
        WriteLine($"{header} start process({data})")
        Await Task.Delay(300)
        WriteLine($"{header} end process({data})")
      End If
    End While
    WriteLine($"{header} EXIT")
  End Function

  ' Consumerを定義（その2−GetConsumingEnumerableメソッドを使ってループ）
  Dim consume2 As Func(Of String, Task) =
  Async Function(header)
    ' キューからデータを取り出す（que.IsCompletedがtrueになるまで）
    For Each dat In que.GetConsumingEnumerable()
      WriteLine($"{header} Dequeue({dat}), count={que.Count}")

      ' キューから取り出したデータを使う処理をシミュレート
      WriteLine($"{header} start process({dat})")
      Await Task.Delay(300)
      WriteLine($"{header} end process({dat})")
    Next
    WriteLine($"{header} EXIT")
  End Function

  System.Threading.Thread.Sleep(100)

  ' Consumerの実行を開始（2スレッド）
  Dim consumeTask1 = Task.Run(Function() consume1("  [1]"))
  System.Threading.Thread.Sleep(10)
  Dim consumeTask2 = Task.Run(Function() consume2("    [2]"))

  ' 実行終了を待機
  Task.WaitAll(produceTask, consumeTask1, consumeTask2)

Finally
  que.Dispose()
End Try
WriteLine("COMPLETED")