レッスン
開始
勉強を始める
CodeGym /コース /C# SELF /高度なパターンと Concurrent...

高度なパターンと Concurrent コレクションの特徴

C# SELF
レベル 58, レッスン 3
使用可能

1.「Producer‑Consumer」 + Concurrent

以前、ConcurrentQueue を話した際に Producer‑Consumer パターンに触れました。ここでは複数のプロデューサーとコンシューマーがいる場合や、正しい終了シグナリングについて詳しく見ていきます。

ConcurrentQueue（や他の Concurrent コレクション）の最大の利点は、自分でスレッドセーフを扱ってくれることです。Enqueue や TryDequeue を lock で囲む必要はありません — 複数のスレッドが共有キューを安全に使えます。

例：複数のプロデューサーと複数のコンシューマー

いくつかのワーカースレッドがタスクを生成し、別のスレッド群がそれを処理します。

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

ConcurrentQueue<string> taskQueue = new ConcurrentQueue<string>();
CancellationTokenSource cts = new CancellationTokenSource(); // コンシューマーをキャンセルするため

// プロデューサー用メソッド
void Producer(string name, int count)
{
    for (int i = 0; i < count; i++)
    {
        string task = $"ザダーチャ_{name}_{i}";
        taskQueue.Enqueue(task);
        Console.WriteLine($"[P:{name}] 追加した: {task}");
        Thread.Sleep(10); 
    }
}

// コンシューマー用メソッド
void Consumer(string name)
{
    while (!cts.Token.IsCancellationRequested || taskQueue.Count > 0)
    {
        if (taskQueue.TryDequeue(out string task))
        {
            Console.WriteLine($"[C:{name}] 処理した: {task}");
            Thread.Sleep(20); 
        }
        else
        {
            Thread.Sleep(50); // キューが空なら待つ
        }
    }
    Console.WriteLine($"[C:{name}] 作業を終了しました。");
}

// Main での起動例:
Task.Run(() => Producer("A", 10)); // プロデューサー A
Task.Run(() => Producer("B", 10)); // プロデューサー B
Task.Run(() => Consumer("1"));    // コンシューマー 1
Task.Run(() => Consumer("2"));    // コンシューマー 2

Thread.Sleep(1000); // 作業時間を与える
cts.Cancel();       // コンシューマーに終了を指示
Thread.Sleep(500); // コンシューマーが残りを処理して終了する時間を与える
C++

ここでは複数のプロデューサーとコンシューマーが同じ ConcurrentQueue をデータ競合なしに使っています：Enqueue と TryDequeue はアトミックです。

終了シグナルの重要性（CancellationTokenSource）

CancellationTokenSource（変数名 cts）を使ってコンシューマーに終了を通知します。これは Producer‑Consumer パターンで重要です：

  • プロデューサーが作業を終える。 要素の追加が終わったら、コンシューマーは無限に空のキューを待つべきではありません。
  • アプリケーションが終了する。 コンシューマーを正しく止める必要があります。

CancellationTokenSource と CancellationToken は標準的な仕組みを提供します：コンシューマーは定期的に IsCancellationRequested をチェックし、必要なら ThrowIfCancellationRequested() を呼びます。

2. BlockingCollection<T>

ConcurrentQueue<T> は Producer‑Consumer に適していますが、空のキューでの待ちや終了のシグナリングは手動でやる必要があります。.NET にはもっと便利な BlockingCollection<T> があります — これは独立したコレクションではなく、任意の IProducerConsumerCollection<T>（例えば ConcurrentQueue） をラップするものです。

BlockingCollection の利点：

  • ブロッキング操作。 Add()/Take() はコレクションが満杯/空ならスレッドをブロックします。IsEmpty を手動でチェックする必要がありません。
  • サイズ制限。 Capacity を指定でき、Add() は制限に達するとブロックされます — メモリ管理に便利です。
  • 終了処理が簡単。 CompleteAdding() は追加の終了を通知し、GetConsumingEnumerable() によってコンシューマーは完全終了まで要素を処理できます。

例：BlockingCollection<T> を使った Producer‑Consumer

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

// BlockingCollection はデフォルトで ConcurrentQueue を使う
BlockingCollection<int> numbers = new BlockingCollection<int>(capacity: 10); // 容量 10 のキュー

void ProducerBC(int count)
{
    for (int i = 0; i < count; i++)
    {
        numbers.Add(i); // キューが満杯ならブロックされる
        Console.WriteLine($"[P] 追加した: {i}");
        Thread.Sleep(50);
    }
    numbers.CompleteAdding(); // プロデューサーが終了したことを通知
    Console.WriteLine("[P] プロデューサーは追加を完了しました。");
}

void ConsumerBC()
{
    // GetConsumingEnumerable は要素があるか CompleteAdding が呼ばれるまでブロックする
    foreach (var item in numbers.GetConsumingEnumerable())
    {
        Console.WriteLine($"[C] 処理した: {item}");
        Thread.Sleep(100);
    }
    Console.WriteLine("[C] コンシューマーは作業を終了しました。");
}

// Main での起動例:
Task producerTask = Task.Run(() => ProducerBC(15)); // 15 要素、容量は 10
Task consumerTask = Task.Run(ConsumerBC);
Task.WaitAll(producerTask, consumerTask); // 終了を待つ

GetConsumingEnumerable() によってコンシューマー側のコードがかなりすっきりする点に注意してください。ブロッキング操作やサイズ制限が必要なら BlockingCollection はとても有用です。

3. Concurrent コレクションの追加メソッドとプロパティ

IsEmpty, Count

  • IsEmpty (bool): コレクションが空かどうか。
  • Count (int): 現在の要素数。

例：IsEmpty と Count の使用

using System.Collections.Concurrent;

ConcurrentQueue<string> q = new ConcurrentQueue<string>();
Console.WriteLine($"キューは空ですか？ {q.IsEmpty}"); // True

q.Enqueue("A");
q.Enqueue("B");
Console.WriteLine($"キューの要素数: {q.Count}"); // 2
Console.WriteLine($"キューは空ですか？ {q.IsEmpty}"); // False

q.TryDequeue(out var itemA);
Console.WriteLine($"キューの要素数: {q.Count}"); // 1

配列への変換（ToArray()）

すべての Concurrent コレクションは ToArray() メソッドを提供しており、要素のスナップショットを返します。

例：ToArray() の使用

using System.Collections.Concurrent;

ConcurrentStack<int> s = new ConcurrentStack<int>();
s.Push(10);
s.Push(20);
s.Push(30);

int[] items = s.ToArray(); // 新しい配列を作成: [30, 20, 10] (スタックでは LIFO)
Console.WriteLine($"配列の要素: {string.Join(", ", items)}");

// コレクション自体は変わらない
Console.WriteLine($"ToArray 後のスタックの要素数: {s.Count}"); // 3

コレクションのクリア

.NET 6+ では、多くの Concurrent コレクションに全要素を削除するための Clear() メソッドが追加されました。

例：コレクションのクリア

using System.Collections.Concurrent;

ConcurrentBag<string> bag = new ConcurrentBag<string>();
bag.Add("Alpha");
bag.Add("Beta");
Console.WriteLine($"Bag の要素数: {bag.Count}"); // 2

bag.Clear(); // コレクションをクリア
Console.WriteLine($"クリア後の Bag の要素数: {bag.Count}"); // 0
Console.WriteLine($"Bag は空ですか？ {bag.IsEmpty}"); // True

4. Concurrent コレクションの振る舞いの注意点

「スナップショット」的な振る舞いを忘れないでください。Count や ToArray() の結果は、コレクションのある瞬間の状態を反映します。並行して変更が行われている場合、その直後にはすぐ古くなる可能性があります。

例：Count と ToArray() — 「スナップショット」

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

ConcurrentQueue<int> snapshotQueue = new ConcurrentQueue<int>();

void AddItemsContinuously()
{
    for (int i = 0; i < 1000; i++)
    {
        snapshotQueue.Enqueue(i);
        Thread.Sleep(1); 
    }
}

// Main での起動例:
Task.Run(AddItemsContinuously); // 要素を継続的に追加するスレッド

Thread.Sleep(100); // 少し追加される時間を与える
Console.WriteLine($"現在の Count: {snapshotQueue.Count}"); // 50、80、120... のどれか
Thread.Sleep(100);
Console.WriteLine($"現在の Count（再度）: {snapshotQueue.Count}"); // 別の値になる
int[] currentItems = snapshotQueue.ToArray();
Console.WriteLine($"ToArray() の要素数: {currentItems.Length}"); // 最後の Count と異なる可能性あり

アクティブに変化している間は Count を厳密な要素数の保証として信用しないでください。

5. Concurrent コレクションのイテレーションの注意点

個々の操作（Add、TryTake、TryPop、GetOrAdd など）はスレッドセーフです。しかし、他のスレッドが並行してコレクションを変更しているときの foreach によるイテレーションは、すべての要素を見ることを保証しないし、予期せぬ抜けや重複が起こる可能性があります。

例：変更中のイテレーション

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

ConcurrentQueue<int> iterQueue = new ConcurrentQueue<int>();

// 初期要素を追加
for (int i = 0; i < 10; i++) iterQueue.Enqueue(i);

// 変更スレッド
void Modifier()
{
    for (int i = 10; i < 20; i++)
    {
        iterQueue.Enqueue(i); // 新しい要素を追加
        Thread.Sleep(50);
    }
}

// イテレータスレッド
void Iterator()
{
    Console.WriteLine("イテレーションを開始します...");
    int count = 0;
    foreach (var item in iterQueue) // イテレーション
    {
        Console.Write($"{item} ");
        count++;
        Thread.Sleep(30); // 処理をシミュレートして Modifier にコレクションを変更させる
    }
    Console.WriteLine($"\nイテレーション終了。読み取った要素数: {count}。");
    Console.WriteLine($"現在のキューの要素数: {iterQueue.Count}");
}

// Main での起動例:
Task.Run(Modifier);
Task.Run(Iterator);
Thread.Sleep(1500); // 作業時間を与える

ルール： 固定された要素セット（例えばレポート）を扱う必要があるなら、まず ToArray() で「スナップショット」を取り、それをイテレートしてください： 

// コレクションが変化する可能性がある場合の正しいイテレーション方法
int[] snapshot = iterQueue.ToArray();
foreach (var item in snapshot)
{
    // ここでは不変の配列スナップショットをイテレートする
}

これで、Concurrent コレクションの高度なパターンと特徴の紹介を終わります：複数参加者の Producer‑Consumer や BlockingCollection、そして Count、ToArray()、イテレーションに関する重要な注意点を見てきました。

2
タスク
C# SELF, レベル 58, レッスン 3
ロック未解除
BlockingCollectionのサイズ制限付きサンプル
BlockingCollectionのサイズ制限付きサンプル
オープン
コメント
TO VIEW ALL COMMENTS OR TO MAKE A COMMENT,
GO TO FULL VERSION