Concurrent 컬렉션의 고급 패턴과 특징 Concurrent-컬렉션

1. 생산자‑소비자 패턴 + Concurrent

우리는 이미 ConcurrentQueue를 이야기하면서 생산자‑소비자 패턴을 잠깐 봤어. 여기선 여러 생산자와 여러 소비자가 있을 때, 그리고 올바른 종료 신호를 다루는 방법을 좀 더 자세히 살펴볼게.

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); // 소비자들이 남은 작업을 처리하고 끝낼 시간

여기선 여러 생산자와 소비자가 동일한 ConcurrentQueue로 동시에 작업하면서 데이터 레이스 없이 동작해: Enqueue와 TryDequeue는 원자적이야.

종료 신호의 중요성 (CancellationTokenSource)

우린 소비자에게 종료를 알리기 위해 CancellationTokenSource (cts)를 사용해. 이건 Producer‑Consumer 패턴에서 아주 중요해:

• 생산자들이 작업을 모두 끝냈을 때. 더 이상 요소를 추가하지 않으면 소비자가 무한히 빈 큐를 기다리면 안 돼.
• 애플리케이션 종료 시. 소비자들을 올바르게 멈춰야 해.

CancellationTokenSource와 CancellationToken은 표준 메커니즘을 제공해: 소비자는 주기적으로 IsCancellationRequested를 확인하고 필요하면 ThrowIfCancellationRequested()를 호출할 수 있어.

2. BlockingCollection<T>

ConcurrentQueue<T>는 생산자‑소비자에 잘 맞지만, 비어있을 때 기다리는 로직과 종료 신호를 수동으로 처리해야 해. 좀 더 편하게 하려면 .NET의 BlockingCollection<T>을 쓰면 좋아 — 이건 독립된 컬렉션이라기보다 IProducerConsumerCollection<T> (예: ConcurrentQueue) 위에 얹는 래퍼야.

BlockingCollection의 장점:

• 블로킹 연산. Add()/Take()는 컬렉션이 가득/비었을 때 스레드를 블록해. IsEmpty를 수동으로 검사할 필요가 없어.
• 크기 제한. Capacity를 지정할 수 있어서 메모리 제어에 유용해. Add()는 한계에 도달하면 블록된다.
• 편한 종료 처리. CompleteAdding()로 추가 종료를 알리고, GetConsumingEnumerable()로 소비자는 끝까지 안전하게 처리할 수 있어.

예제: BlockingCollection을 사용한 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); // 작업을 흉내내며 수정자가 컬렉션을 변경할 시간 제공
    }
    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(), 이터레이션 관련 중요한 주의점을 다뤘어.