큐와 스택 Producer-Consumer

1. Введение

기본부터 시작해볼게요. 큐는 기본적인 자료구조로, FIFO (First-In, First-Out) 원칙으로 동작해요. 즉 "먼저 들어온 게 먼저 나간다". 슈퍼마켓의 대기열을 떠올리면 이해하기 쉬워요: 먼저 줄 선 사람이 먼저 서비스 받죠.

멀티스레드 프로그래밍에서 패턴 Производитель-Потребитель (Producer-Consumer)은 가장 흔하고 강력한 패턴 중 하나예요.

• 생산자 (Producers) — 데이터를 생성하거나 작업을 만들어 공용 큐에 넣는 스레드나 애플리케이션의 부분이에요. 이들은 작업을 "생산"합니다.
• 소비자 (Consumers) — 큐에서 데이터를 꺼내 처리하는 스레드나 애플리케이션의 부분이에요. 이들은 작업을 "소비"합니다.

이 패턴은 데이터 흐름을 관리하고 컴포넌트들을 느슨하게 결합하며(생산자가 누가 어떻게 처리할지 알 필요가 없음), 애플리케이션을 더 반응성 있게 만들고 스레드 간에 부하를 고르게 분산하는 데 도움을 줍니다.

Пример: ConcurrentQueue — добавление и извлечение

Как добавлять и извлекать элементы из ConcurrentQueue<T>.

using System.Collections.Concurrent;

ConcurrentQueue<string> tasks = new ConcurrentQueue<string>();

// 항목 추가 (생산자)
tasks.Enqueue("파일 다운로드");
tasks.Enqueue("이미지 처리");
Console.WriteLine($"큐에 작업 수: {tasks.Count}"); // 출력: 큐에 작업 수: 2

// 항목 추출 (소비자)
if (tasks.TryDequeue(out string task1))
{
    Console.WriteLine($"완료된 작업: {task1}"); // 출력: 완료된 작업: 파일 다운로드
}

if (tasks.TryDequeue(out string task2))
{
    Console.WriteLine($"완료된 작업: {task2}"); // 출력: 완료된 작업: 이미지 처리
}

if (!tasks.TryDequeue(out string emptyTask))
{
    Console.WriteLine("큐가 비어 있습니다. 새로운 작업이 없습니다."); // 출력: 큐가 비어 있습니다. 새로운 작업이 없습니다.
}

Основы работы: Enqueue(), TryDequeue()

Enqueue(T item): 요소를 큐의 끝에 추가할 때 사용해요. 이 연산은 스레드-안전합니다. 예를 들어 10개의 서로 다른 스레드에서 동시에 Enqueue를 호출해도 모든 요소가 올바르게 추가됩니다.

TryDequeue(out T item): 큐의 앞에서 요소를 꺼내려고 시도할 때 사용합니다. 소비자에게 중요한 메서드예요. 요소를 성공적으로 꺼내면 true를 반환하고(값은 출력 매개변수 item에 들어감), 큐가 비어 있으면 false를 반환합니다. 중요한 점은 TryDequeue가 큐가 비어 있을 때 스레드를 블로킹하지 않는다는 거예요.

2. Важность TryDequeue() и атомарность операций

TryDequeue()는 단순히 편리한 메서드가 아니에요; 올바른 스레드-안전 동작을 위해 매우 중요합니다. 이 메서드는 원자적입니다: 큐가 비어있는지 확인하고 실제로 요소를 꺼내는 동작이 하나의 분리할 수 없는 연산으로 수행됩니다.

만약 별도의 메서드인 IsEmpty (큐가 비었는지 확인)와 Dequeue (요소 꺼내기)가 있었다면, 두 호출 사이에 다른 스레드가 큐를 비울 수 있어요. 결과적으로 당신의 Dequeue는 예외를 던지거나 잘못된 데이터를 반환할 수 있습니다. TryDequeue는 그런 상황을 완전히 방지합니다.

Пример: Producer-Consumer с несколькими потоками

여기서는 두 개의 생산자 스레드와 하나의 소비자 스레드를 실행하는 예제입니다.

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

ConcurrentQueue<int> dataQueue = new ConcurrentQueue<int>();
bool producersDone = false; // 소비자에게 신호를 보내는 플래그

void Producer(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        dataQueue.Enqueue(start + i);
        Console.WriteLine($"[생] 추가: {start + i}");
        Thread.Sleep(10); 
    }
}

void Consumer()
{
    while (!producersDone || dataQueue.Count > 0) // 데이터가 있거나 생산자들이 아직 작업 중이면 계속함
    {
        if (dataQueue.TryDequeue(out int item))
        {
            Console.WriteLine($"[소] 처리: {item}");
        }
        else
        {
            Thread.Sleep(50); // 큐가 비어 있으면 잠시 기다림
        }
    }
    Console.WriteLine("[소] 작업을 마쳤습니다.");
}

// Main에서 예제 실행:
// Task.Run(() => Producer(1, 5));
// Task.Run(() => Producer(100, 5)); // 두 번째 생산자
// Task.Run(() => Consumer());
// Thread.Sleep(600); // 스레드들이 작업할 시간을 줌
// producersDone = true; // 생산자들이 끝났다고 신호
// Thread.Sleep(200); // 소비자가 남은 걸 처리할 시간을 줌

이 간단한 예제에서는 플래그 producersDone와 Thread.Sleep을 사용해 종료를 흉내 냈다는 점을 유의하세요. 실제 애플리케이션에서는 종료 동기화를 위해 보통 CancellationTokenSource나 BlockingCollection<T> 같은 더 신뢰성 있는 수단을 사용합니다.

ConcurrentQueue<T>는 다음 시나리오에 적합합니다:

• 요소 처리 순서가 중요할 때 (FIFO).
• 여러 스레드가 요소를 추가하거나, 여러 스레드가 요소를 꺼낼 때.
• 수동 락 관리를 하지 않고도 높은 성능이 필요할 때.

3. Стек для producer-consumer (LIFO)

스택은 다른 기본 자료구조로, LIFO (Last-In, First-Out) 원칙으로 동작해요. 즉 "나중에 들어온 것이 먼저 나간다". 접시 더미를 떠올리면, 항상 맨 위의 접시를 집고 새 접시는 맨 위에 올려놓죠.

ConcurrentStack<T>는 ConcurrentQueue<T>와 마찬가지로 스레드-안전하며, producer-consumer 패턴에서 역순의 처리 순서가 필요할 때 사용할 수 있습니다.

Пример: ConcurrentStack — добавление и извлечение

using System.Collections.Concurrent;

ConcurrentStack<string> commandStack = new ConcurrentStack<string>();

// 명령 추가 (생산자)
commandStack.Push("텍스트 선택");
commandStack.Push("글꼴 변경");
commandStack.Push("문서 저장");
Console.WriteLine($"스택의 명령 수: {commandStack.Count}"); // 출력: 스택의 명령 수: 3

// 명령 추출 (소비자)
if (commandStack.TryPop(out string cmd1))
{
    Console.WriteLine($"취소된 명령: {cmd1}"); // 출력: 취소된 명령: 문서 저장
}

if (commandStack.TryPop(out string cmd2))
{
    Console.WriteLine($"취소된 명령: {cmd2}"); // 출력: 취소된 명령: 글꼴 변경
}

if (!commandStack.TryPop(out string emptyCmd))
{
    Console.WriteLine("명령 스택이 비어 있습니다."); // 출력: 명령 스택이 비어 있습니다.
}

4. Основы работы: Push(), TryPop()

Push(T item): 요소를 스택의 꼭대기에 추가할 때 사용합니다. 이 연산은 스레드-안전합니다.

TryPop(out T item): 스택의 꼭대기에서 요소를 꺼내려고 시도할 때 사용합니다. 요소를 성공적으로 꺼내면 true를 반환하고, 스택이 비어 있으면 false를 반환합니다. TryDequeue처럼 이 연산도 원자적이라 레이스 컨디션을 방지합니다.

Пример: использование ConcurrentStack для пула объектов

스택은 객체 풀 구현에 아주 잘 맞습니다: 빼서 쓰고, 다시 넣고.

using System.Collections.Concurrent;

class Connection { /* 간단한 스텁 */ public Guid Id { get; } = Guid.NewGuid(); }

ConcurrentStack<Connection> connectionPool = new ConcurrentStack<Connection>();

// 초기 연결로 풀을 채움
for (int i = 0; i < 3; i++)
{
    connectionPool.Push(new Connection());
}
Console.WriteLine($"풀의 연결 수: {connectionPool.Count}"); // 출력: 풀의 연결 수: 3

void UseConnection()
{
    if (connectionPool.TryPop(out Connection conn))
    {
        Console.WriteLine($"[풀] 사용된 연결: {conn.Id}");
        // 연결 사용 흉내
        Thread.Sleep(50); 
        connectionPool.Push(conn); // 풀에 반환
        Console.WriteLine($"[풀] 연결 반환: {conn.Id}. 풀에: {connectionPool.Count}");
    }
    else
    {
        Console.WriteLine("[풀] 연결 풀이 비어 있습니다. 새로 생성합니다.");
        // 보통 풀 비었을 때 새 연결을 생성함
        connectionPool.Push(new Connection()); 
    }
}

// Main에서 예제 실행:
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Thread.Sleep(500);

이 예제에서는 여러 스레드가 안전하게 공유 풀에서 연결을 가져오고 돌려줄 수 있다는 것을 보여줍니다.

5. Примеры применения и сравнение с ConcurrentQueue

ConcurrentStack<T>는 다음과 같은 경우에 사용합니다:

• 처리 순서가 LIFO로 중요할 때 (예: "실행 취소" 기능을 위한 작업 기록).
• 가장 최근에 추가된 요소에 빠르게 접근해야 할 때 (종종 CPU 캐시에 '핫'하게 남아 있음).
• 스택 기반 알고리즘을 구현할 때 (깊이 우선 탐색, 표현식 파싱 등).

Сравнение и выбор подходящей коллекции

ConcurrentQueue

ConcurrentStack

ConcurrentQueue와 ConcurrentStack 중 무엇을 쓸지는 producer-consumer 시나리오에서 요구되는 처리 순서에 달려 있어요. 두 컬렉션 모두 높은 성능과 스레드-안전성을 "기본 제공"으로 제공하므로 수동으로 락을 관리할 필요를 줄여주고 확장 가능한 멀티스레드 시스템을 만드는 데 도움을 줍니다.