다중 스레드 요청 처리 시 발생하는 예외 상황

시스템 내에서 외부 호출을 수행하고 그 결과를 바탕으로 특정 이벤트가 실행되도록 설계되는 경우, 이러한 요청 정보를 일시적으로 저장할 수 있는 컨테이너가 필요합니다. 초기 버전에서는 단순한 리스트(List) 타입을 사용하여 요청을 적재하고 조건이 충족되면 발송하는 방식을 적용했습니다. 그러나 여러 스레드에서 동시에 해당 리스트에 접근할 때 불안정성이 발견되었습니다.

예를 들어, 한 스레드는 리스트를 순회하며 데이터를 출력하는 동안 다른 스레드가 리스트 인덱스를 제거하거나 수정하려는 시도입니다. 이 경우 열거 (Enumeration) 와 삭제 연산이 충돌하여 런타임 예외가 발생합니다.

<span style="color: blue;">using</span> System;
<span style="color: blue;">using</span> System.Threading;
<span style="color: blue;">namespace</span> ConcurrentSafetyDemo
{
    <span style="color: blue;">class</span> Program
    {
        static <span style="color: blue;">void</span> Main(<span style="color: blue;">string</span>[] args)
        {
            <span style="color: #4d4d4d;">// 원본 데이터 리스트 생성</span>
            var sharedItems = <span style="color: blue;">new</span> List<<span style="color: blue;">int</span>>();
            <span style="color: blue;">for</span> (<span style="color: blue;">int</span> id = 0; id &lt; 10; id++)
            {
                sharedItems.Add(id);
            }

            <span style="color: blue;">var</span> readerThread = <span style="color: blue;">new</span> Thread(() =&gt;
            {
                <span style="color: blue;">foreach</span> (<span style="color: blue;">var</span> data <span style="color: blue;">in</span> sharedItems)
                {
                    Console.WriteLine($"읽는 스레드 ID: {data}");
                    Thread.Sleep(1000);
                }
            });
            readerThread.Start();

            <span style="color: blue;">var</span> writerThread = <span style="color: blue;">new</span> Thread(() =&gt;
            {
                Thread.Sleep(1000);
                <span style="color: #4d4d4d;">// 순차적 제거 작업</span>
                sharedItems.RemoveAt(1);
                sharedItems.RemoveAt(2); 
                
                <span style="color: blue;">foreach</span> (<span style="color: blue;">var</span> data <span style="color: blue;">in</span> sharedItems)
                {
                    Console.WriteLine($"수정 후 확인: {data}");
                }
            });
            writerThread.Start();
            
            Thread.Sleep(5000);
        }
    }
}

위 코드를 실행하면 InvalidOperationException 가 발생하며 메시지는 "컬렉션이 변경되었습니다"라고 표시됩니다. 이는 읽기 스레드가 열거를 진행 중인 도중 쓰기 스레드가 구조 변화를 일으켰기 때문입니다. 이를 방지하기 위해서는 공유 자원 접근 시 동기화를 보장해야 합니다.

락 (Lock) 을 통한 데이터 일관성 유지

문제를 해결하기 위해 공인된 객체 (Lock Object) 를 정의하고 모든 읽기/쓰기 연산을 감싸주는 방식이 적용됩니다. 이렇게 하면 열거가 진행되는 동안 수정 작업이 대기하게 되어 안전성이 확보됩니다.

<span style="color: blue;">namespace</span> ConcurrentSafetyDemo
{
    <span style="color: blue;">class</span> Program
    {
        static <span style="color: blue;">object</span> syncRoot = <span style="color: blue;">new</span> object();

        static <span style="color: blue;">void</span> Main(<span style="color: blue;">string</span>[] args)
        {
            <span style="color: blue;">var</span> sharedItems = <span style="color: blue;">new</span> List<<span style="color: blue;">int</span>>();
            <span style="color: blue;">for</span> (<span style="color: blue;">int</span> i = 0; i &lt; 10; i++) sharedItems.Add(i);

            <span style="color: blue;">var</span> t1 = <span style="color: blue;">new</span> Thread(() =&gt;
            {
                <span style="color: blue;">lock</span> (syncRoot)
                {
                    <span style="color: blue;">foreach</span> (<span style="color: blue;">var</span> item <span style="color: blue;">in</span> sharedItems)
                    {
                        Console.WriteLine(t1.ManagedThreadId + "-" + item);
                        Thread.Sleep(500);
                    }
                }
            });
            t1.Start();

            <span style="color: blue;">var</span> t2 = <span style="color: blue;">new</span> Thread(() =&gt;
            {
                Thread.Sleep(1000);
                <span style="color: blue;">lock</span> (syncRoot)
                {
                    sharedItems.RemoveRange(0, 2);
                    Console.WriteLine($"삭제 후 크기: {sharedItems.Count}");
                }
            });
            t2.Start();
        }
    }
}

이를 통해 기본 컬렉션도 락을 사용하면 안전하게 사용할 수 있습니다. 하지만 반복적인 잠금 해제/획득은 성능 오버헤드를 유발할 수 있으므로, 더 적합한 자료구조를 검토해 볼 수 있습니다. 특히 생산자와 소비자 (Producer-Consumer) 패턴에서는 큐 (Queue) 가 자주 사용됩니다.

큐 (Queue) 의 다중 스레드 접근 한계

일반적인 Queue 클래스는 스레드 안전을 보장하지 않습니다. 만약 무작위로 요소가 추가되거나 제거될 경우 동일한 예외가 발생할 수 있습니다. 우선 기본적인 큐 구조로 테스트 해보겠습니다.

<span style="color: blue;">var</span> workQueue = <span style="color: blue;">new</span> Queue<<span style="color: blue;">int</span>>();
<span style="color: blue;">var</span> counter = 0;

<span style="color: blue;">Task</span>.Factory.StartNew(() =&gt;
{
    <span style="color: blue;">while</span> (true)
    {
        Thread.Sleep(500);
        Console.WriteLine("생성: " + counter);
        workQueue.Enqueue(counter);
        counter++;
    }
});

<span style="color: blue;">Task</span>.Factory.StartNew(() =&gt;
{
    <span style="color: blue;">while</span> (true)
    {
        <span style="color: blue;">if</span> (workQueue.Count &gt; 0)
        {
            <span style="color: blue;">var</span> val = workQueue.Dequeue();
            Console.WriteLine("소비됨: " + val);
        }
    }
});

위 코드처럼 단순한 카운터 증가나 Dequeue 를 사용할 때는 내부적인 동기화가 빠져 있어 데이터 손실이나 예외 가능성이 있습니다. 이를 개선하기 위해 System.Collections.Concurrent 네임스페이스의 ConcurrentQueue 를 도입합니다.

동시성 큐 (ConcurrentQueue) 활용

ConcurrentQueue 는 내부적으로 락 없이도 스레드 안전을 제공합니다. 하지만 단순히 Count > 0 확인보다는 TryDequeue 메서드를 사용하여 대기 시간을 줄이는 것이 좋습니다. 다만 여전히 루프 내에 Try를 계속 실행해야 하는 번거로움이 있습니다.

<span style="color: blue;">var</span> queueStore = <span style="color: blue;">new</span> ConcurrentQueue<<span style="color: blue;">int</span>>();
<span style="color: blue;">var</span> idx = 0;

<span style="color: blue;">Task</span>.Factory.StartNew(() =&gt;
{
    <span style="color: blue;">while</span> (true)
    {
        queueStore.Enqueue(idx++);
    }
});

<span style="color: blue;">Task</span>.Factory.StartNew(() =&gt;
{
    <span style="color: blue;">while</span> (true)
    {
        <span style="color: blue;">int</span> result;
        <span style="color: blue;">if</span> (queueStore.TryDequeue(out result))
        {
            Console.WriteLine(Console.CurrentThread.ManagedThreadId + $" 수신: {result}");
        }
    }
});

이 방법은 효율적이지만, 큐가 비어있을 때 CPU 소모 없이 기다리는 (Block 하는) 기능이 내장되어 있지 않습니다. 매번 체크하면서 자원을 낭비하는 Polling 방식의 단점이 있을 수 있습니다.

블로킹 컬렉션 (BlockingCollection) 으로 완전한 해결

더 세련되고 효율적인 해결책으로 BlockingCollection 을 사용합니다. 이는 이미 블로킹 동작과 IEnumerable 인터페이스를 지원하므로, 항목이 없을 때 자동으로 스레드를 잠자기 상태로 만들거나 아이템이 들어올 때까지 기다리게 할 수 있습니다.

<span style="color: blue;">var</span> boundedQueue = <span style="color: blue;">new</span> BlockingCollection<<span style="color: blue;">int</span>>(new ArrayBlockingQueue<<span style="color: blue;">int</span>>(10));

<span style="color: blue;">var</span> taskProducer = <span style="color: blue;">Task</span>.Factory.StartNew(() =&gt;
{
    <span style="color: blue;">for</span> (<span style="color: blue;">int</span> i = 0; i &lt; 20; i++)
    {
        boundedQueue.Add(i);
        Console.WriteLine("생산 완료: " + i);
    }
    boundedQueue.CompleteAdding();
});

<span style="color: blue;">var</span> taskConsumerA = <span style="color: blue;">Task</span>.Factory.StartNew(() =&gt;
{
    <span style="color: blue;">foreach</span> (<span style="color: blue;">var</span> v <span style="color: blue;">in</span> boundedQueue.GetConsumingEnumerable())
    {
        Console.WriteLine("워크러 A 처리: " + v);
    }
});

<span style="color: blue;">var</span> taskConsumerB = <span style="color: blue;">Task</span>.Factory.StartNew(() =&gt;
{
    <span style="color: blue;">foreach</span> (<span style="color: blue;">var</span> v <span style="color: blue;">in</span> boundedQueue.GetConsumingEnumerable())
    {
        <span style="color: blue;">if</span> (v % 2 == 1)
        {
            <span style="color: #4d4d4d;">// 홀수면 다시 큐로 보냄</span>
            boundedQueue.Add(v * 10);
        }
    }
});

<span style="color: blue;">Task</span>.Wait(taskProducer, taskConsumerA, taskConsumerB);

위와 같이 GetConsumingEnumerable 을 사용하면 내부적으로 블로킹이 일어나므로polling 코드가 필요없습니다. 프로듀서가 CompleteAdding 을 호출하여 데이터 전송을 중단하면 소비자들 역시 정상적으로 종료될 수 있습니다.