동시성 - lock·Interlocked·Channel

이 장에서 배우는 것

택배 센터의 분류 서비스는 여러 스캐너와 여러 작업자가 같은 집계 값을 동시에 갱신한다. 코드는 한 줄씩 읽으면 맞아 보이지만, 스레드 여러 개가 같은 메모리를 건드리면 결과가 실행마다 달라진다. 이 장에서는 그런 오류를 결정적으로 재현해 보고, 상황에 맞는 도구를 고르는 기준을 세운다. 앞 장에서 다룬 취소 토큰과 비동기 흐름을 알고 있다고 가정한다.

• 경쟁 상태(race condition)가 생기는 이유를 읽기-수정-쓰기 순서로 설명하고 재현할 수 있다.
• lock 과 Interlocked 의 보호 범위 차이를 알고 알맞은 쪽을 고른다.
• 동시 컬렉션이 보장하는 것과 보장하지 않는 것을 구분한다.
• Channel<T> 로 생산자-소비자 구조를 만들고 정상 종료까지 처리한다.
• Parallel.ForEachAsync 로 동시에 실행할 작업 수를 제한한다.

문제 상황

분류 센터에는 게이트가 여러 개 있고, 게이트마다 스캐너가 소포를 읽을 때마다 "오늘 처리한 건수"를 올린다. 처음에는 필드 하나에 count++ 를 쓴다. 테스트 서버에서 스캐너 한 대로 돌리면 건수가 정확하다. 운영에서 스캐너 여러 대가 붙자 화면의 건수가 실제 스캔 로그보다 조금씩 적게 나온다. 어제는 12건, 오늘은 40건이 모자란다. 재실행하면 숫자가 또 다르다.

원인은 count++ 가 한 동작처럼 보여도 실제로는 세 단계라는 데 있다. 값을 읽고, 1을 더하고, 결과를 쓴다. 두 스레드가 같은 값을 읽은 뒤 각자 써 버리면 증가분 하나가 사라진다. 이렇게 실행 순서에 따라 결과가 달라지는 오류를 경쟁 상태라고 한다. 이런 오류는 재현이 어렵다는 점이 더 골치 아프다. 그래서 이 장은 먼저 스레드 실행 순서를 억지로 맞춰 오류를 매번 똑같이 재현한다.

이어서 같은 집계를 lock, Interlocked, 동시 컬렉션으로 고친다. 마지막으로 접수 쪽과 처리 쪽을 Channel 로 분리하고, 배송 조회 같은 대기 작업을 Parallel.ForEachAsync 로 병렬 처리한다.

경쟁 상태의 구조

읽기-수정-쓰기는 나뉘어 실행된다

공유 필드 _scanned 에 1을 더하는 코드는 CPU 수준에서 대략 다음 세 동작이다. 스레드 전환은 이 사이 어디서든 일어날 수 있다.

1. 필드 값을 지역 레지스터로 읽는다.
2. 레지스터 값에 1을 더한다.
3. 결과를 필드에 쓴다.

스레드 A와 B가 모두 1단계를 끝낸 뒤 3단계를 하면 둘 다 0을 읽고 둘 다 1을 쓴다. 스캔은 두 번 일어났지만 집계는 1이다. 아래 그림이 이 순서를 보여 준다.

실제 서비스에서는 이런 겹침이 드물게 일어나서 오류가 간헐적으로 보인다. 예제에서는 Barrier 로 "두 스레드가 모두 읽을 때까지 쓰기를 미룬다"는 조건을 만들어 겹침을 항상 일으킨다. Barrier 는 재현용 장치일 뿐 운영 코드에 넣을 것이 아니다.

동기화 도구를 고르는 기준

보호해야 하는 것이 무엇이냐에 따라 도구가 갈린다. 변수 하나의 단순 갱신이면 Interlocked 가 가볍다. 여러 값이 함께 맞아야 하거나 컬렉션을 고치는 구간이면 lock 이 맞다. 아래 표에 정리했다.

동기화 도구별 보호 대상과 쓸 곳

도구	보호 대상	대기 방식	쓸 곳
Interlocked	변수 하나의 원자적 갱신	대기 없음(재시도)	건수, 합계, 최댓값
lock	여러 문장으로 된 구간	스레드를 막고 기다림	딕셔너리와 불변식 유지
SemaphoreSlim	구간 또는 동시 실행 수	await 로 기다림	구간 안에서 비동기 호출이 필요할 때
Channel<T>	작업 전달 자체	await 로 기다림	생산자와 소비자 분리

lock 과 Interlocked

lock: 한 번에 한 스레드만 들어가는 구간

lock (객체) { ... } 블록은 같은 객체로 잠근 다른 스레드가 블록에 들어오지 못하게 한다. .NET 9부터는 전용 형식인 System.Threading.Lock 이 있어서 private readonly Lock _lock = new(); 로 잠금 객체를 만든다. lock 문에 이 형식을 쓰면 컴파일러가 전용 경로로 처리한다. 이전 방식인 object 잠금도 동작하지만 새 코드는 Lock 을 쓰는 편이 의도가 분명하다.

잠금 객체는 클래스 안에 숨겨 두는 것이 원칙이다. this, 문자열, typeof(...) 같은 바깥에서도 접근할 수 있는 객체를 잠그면 다른 코드가 같은 객체로 잠가 뜻하지 않게 서로 기다리게 된다. 잠금 구간은 짧을수록 좋다. 구간 안에서 파일이나 네트워크를 호출하면 그동안 다른 스레드가 모두 멈춘다.

lock 블록 안에서는 await 를 쓸 수 없다. 잠금은 스레드에 묶여 있는데 await 뒤에는 다른 스레드에서 이어질 수 있기 때문이다. 컴파일러가 오류로 막아 준다. 구간 안에서 비동기 호출이 꼭 필요하면 SemaphoreSlim 의 WaitAsync 를 쓴다.

Interlocked: 하드웨어 원자 연산

Interlocked.Increment 와 Interlocked.Add 는 읽기-수정-쓰기를 쪼개지지 않는 하나의 연산으로 수행한다. 스레드를 막지 않기 때문에 lock 보다 가볍다. 다만 보호되는 것은 그 변수 하나뿐이다. "건수와 합계가 항상 함께 맞아야 한다"는 조건은 Interlocked 두 번으로 지킬 수 없다. 두 호출 사이에 다른 스레드가 값을 읽으면 건수만 올라간 상태를 보게 된다.

최댓값처럼 "지금 값을 보고 조건이 맞으면 바꾼다"는 갱신에는 CompareExchange 를 반복문과 묶는다. 현재 값이 내가 읽은 값 그대로일 때만 새 값을 쓰고, 그 사이 누가 바꿨으면 다시 읽어 재시도한다. 예제의 UpdateMax 가 이 형태다.

동시 컬렉션

System.Collections.Concurrent 의 컬렉션은 개별 연산이 스레드에 안전하도록 만들어져 있다. ConcurrentDictionary 의 AddOrUpdate 는 "없으면 추가하고 있으면 갱신한다"를 한 연산으로 처리한다. 하지만 안전한 연산 두 개를 이어 붙인다고 그 조합이 안전한 것은 아니다. ContainsKey 로 확인한 뒤 [key] = ... 로 쓰면, 그 사이에 다른 스레드가 끼어들 수 있다. 확인과 쓰기를 한 메서드로 처리하는 TryAdd, AddOrUpdate, GetOrAdd 를 써야 한다.

한 가지 더 알아 둘 점이 있다. GetOrAdd 와 AddOrUpdate 에 넘기는 값 계산 함수는 경쟁이 생기면 같은 키에 대해 두 번 이상 호출될 수 있다. 그중 하나의 결과만 저장된다. 이 함수 안에서 외부 상태를 바꾸거나 비용이 큰 작업을 하면 안 된다.

동시 컬렉션 선택 기준

컬렉션	성격	어울리는 용도	주의
ConcurrentDictionary	키-값 조회와 갱신	지역별·상태별 집계, 캐시	값 계산 함수가 중복 호출될 수 있음
ConcurrentQueue	선입선출	작업을 순서대로 넘길 때	비어 있는지 확인 후 꺼내기는 TryDequeue 로
ConcurrentBag	순서 없는 모음	결과를 모아 두기만 할 때	꺼내는 순서를 기대하면 안 됨
Channel<T>	대기와 종료가 있는 통로	생산자-소비자	완료 신호를 직접 보내야 함

Channel 로 생산자와 소비자 나누기

소포 접수는 빠르고 분류 처리는 느릴 수 있다. 접수 쪽이 처리 쪽을 기다리게 하고 싶지 않아서 큐를 하나 두는 것이 생산자-소비자 구조다. Channel 은 이 구조를 async/await 와 어울리게 만든 통로다. 생산자는 Writer.WriteAsync 로 넣고, 소비자는 Reader.ReadAllAsync 로 꺼낸다. 소비자가 여러 명이어도 항목 하나는 한 소비자에게만 전달된다.

채널에는 두 종류가 있다. CreateUnbounded 는 용량 제한이 없어서 생산자가 절대 기다리지 않지만 소비자가 느리면 메모리가 계속 늘어난다. CreateBounded 는 용량을 정하고, 가득 차면 WriteAsync 가 자리가 날 때까지 기다린다. 이 성질을 배압(backpressure)이라 부른다. 처리 속도를 넘는 유입을 막는 안전장치이므로 서비스 코드에서는 유한 채널을 기본으로 삼는다.

종료 처리가 가장 자주 빠진다. 소비자의 ReadAllAsync 반복은 쓰는 쪽이 Writer.Complete() 를 호출해야 끝난다. 호출하지 않으면 소비자가 다음 항목을 계속 기다리므로 프로그램이 끝나지 않는다. 생산자가 여럿이면 모두 끝난 뒤에 한 번만 Complete 를 호출해야 한다. 예제는 Task.WhenAll(producers) 뒤에 호출한다.

Parallel.ForEachAsync

배송 조회처럼 외부 시스템을 기다리는 작업 수백 건을 처리한다고 하자. foreach 로 하나씩 await 하면 총 시간이 합산되고, 전부 Task.WhenAll 로 한꺼번에 시작하면 외부 서버에 동시 요청이 몰린다. Parallel.ForEachAsync 는 ParallelOptions.MaxDegreeOfParallelism 로 동시에 실행할 본문 수의 상한을 정한다. 본문은 (항목, CancellationToken) 을 받는 비동기 람다이고, 앞 장에서 다룬 취소 토큰이 그대로 전달된다.

이 API는 처리 순서를 보장하지 않는다. 결과를 순서대로 써야 하면 항목 번호로 미리 정해 둔 자리에 넣거나, 끝난 뒤에 정렬한다. 본문이 여러 스레드에서 동시에 실행되므로 본문 안에서 공유 상태를 건드리면 이 장의 앞부분 규칙을 그대로 적용해야 한다. CPU를 오래 쓰는 계산이라면 Parallel.For 가 맞고, 기다리는 시간이 대부분인 작업이면 ForEachAsync 가 맞다.

완성 코드

기본 콘솔 앱 템플릿(암시적 using 사용)에서 Program.cs 하나로 실행한다. 출력은 스레드 순서와 무관하게 항상 같다.

using System.Collections.Concurrent;
using System.Threading.Channels;

RaceDemo.Run();
LockDemo.Run();
InterlockedDemo.Run();
ConcurrentDemo.Run();
await ChannelDemo.RunAsync();
await ParallelDemo.RunAsync();

record Parcel(int Id, string Region, int WeightGram);

static class Sites
{
    public static readonly string[] Regions = ["서울", "부산", "대전", "광주"];

    public static void PrintSorted(ConcurrentDictionary<string, int> counts)
    {
        foreach (var pair in counts.OrderBy(p => p.Key, StringComparer.Ordinal))
        {
            Console.WriteLine($"{pair.Key}: {pair.Value}");
        }
    }
}

static class RaceDemo
{
    private static int _scanned;

    public static void Run()
    {
        Console.WriteLine("[1] 경쟁 상태 재현");
        _scanned = 0;
        using var barrier = new Barrier(2);
        var workers = new Thread[2];
        for (int i = 0; i < workers.Length; i++)
        {
            workers[i] = new Thread(() =>
            {
                int seen = _scanned;
                barrier.SignalAndWait();
                _scanned = seen + 1;
            });
            workers[i].Start();
        }
        foreach (var worker in workers)
        {
            worker.Join();
        }
        Console.WriteLine($"스캔 2회, 집계 {_scanned}");
    }
}

static class LockDemo
{
    private sealed class GateCounter
    {
        private readonly Lock _lock = new();
        private readonly Dictionary<string, int> _counts = new();

        public void Add(string gate)
        {
            lock (_lock)
            {
                _counts[gate] = _counts.GetValueOrDefault(gate) + 1;
            }
        }

        public List<KeyValuePair<string, int>> Snapshot()
        {
            lock (_lock)
            {
                return _counts.OrderBy(p => p.Key, StringComparer.Ordinal).ToList();
            }
        }
    }

    public static void Run()
    {
        Console.WriteLine("[2] lock 으로 게이트별 집계");
        var counter = new GateCounter();
        string[] gates = ["A", "B"];
        Parallel.For(0, 4, worker =>
        {
            for (int n = 0; n < 25_000; n++)
            {
                counter.Add(gates[(worker + n) % 2]);
            }
        });

        int total = 0;
        foreach (var pair in counter.Snapshot())
        {
            Console.WriteLine($"{pair.Key}: {pair.Value}");
            total += pair.Value;
        }
        Console.WriteLine($"합계: {total}");
    }
}

static class InterlockedDemo
{
    private static long _totalGram;
    private static int _parcelCount;
    private static int _maxGram;

    public static void UpdateMax(ref int target, int candidate)
    {
        int current = Volatile.Read(ref target);
        while (candidate > current)
        {
            int previous = Interlocked.CompareExchange(ref target, candidate, current);
            if (previous == current)
            {
                return;
            }
            current = previous;
        }
    }

    public static void Run()
    {
        Console.WriteLine("[3] Interlocked 로 통계 집계");
        Parallel.For(1, 1001, id =>
        {
            int weight = 100 + id % 50 * 10;
            Interlocked.Increment(ref _parcelCount);
            Interlocked.Add(ref _totalGram, weight);
            UpdateMax(ref _maxGram, weight);
        });
        Console.WriteLine($"건수: {_parcelCount}");
        Console.WriteLine($"총 무게(g): {_totalGram}");
        Console.WriteLine($"최대 무게(g): {_maxGram}");
    }
}

static class ConcurrentDemo
{
    public static void Run()
    {
        Console.WriteLine("[4] ConcurrentDictionary 로 지역별 집계");
        var byRegion = new ConcurrentDictionary<string, int>();
        Parallel.For(0, 4000, i =>
            byRegion.AddOrUpdate(Sites.Regions[i % 4], 1, (_, old) => old + 1));
        Sites.PrintSorted(byRegion);
    }
}

static class ChannelDemo
{
    public static async Task RunAsync()
    {
        Console.WriteLine("[5] Channel 생산자-소비자");
        var channel = Channel.CreateBounded<Parcel>(new BoundedChannelOptions(8)
        {
            FullMode = BoundedChannelFullMode.Wait
        });

        var producers = Enumerable.Range(0, 2).Select(p => Task.Run(async () =>
        {
            for (int n = 0; n < 500; n++)
            {
                int id = p * 500 + n;
                var parcel = new Parcel(id, Sites.Regions[id % 4], 500 + id % 10 * 100);
                await channel.Writer.WriteAsync(parcel);
            }
        })).ToArray();

        var byRegion = new ConcurrentDictionary<string, int>();
        long totalGram = 0;
        var consumers = Enumerable.Range(0, 3).Select(_ => Task.Run(async () =>
        {
            await foreach (var parcel in channel.Reader.ReadAllAsync())
            {
                byRegion.AddOrUpdate(parcel.Region, 1, (_, old) => old + 1);
                Interlocked.Add(ref totalGram, parcel.WeightGram);
            }
        })).ToArray();

        await Task.WhenAll(producers);
        channel.Writer.Complete();
        await Task.WhenAll(consumers);

        Console.WriteLine($"처리 건수: {byRegion.Values.Sum()}");
        Console.WriteLine($"총 무게(g): {totalGram}");
        Sites.PrintSorted(byRegion);
    }
}

static class ParallelDemo
{
    public static async Task RunAsync()
    {
        Console.WriteLine("[6] Parallel.ForEachAsync 로 배송 조회");
        var status = new ConcurrentDictionary<string, string>();
        int running = 0;
        int peak = 0;
        var options = new ParallelOptions { MaxDegreeOfParallelism = 4 };

        await Parallel.ForEachAsync(Enumerable.Range(1, 20), options, async (n, ct) =>
        {
            int now = Interlocked.Increment(ref running);
            InterlockedDemo.UpdateMax(ref peak, now);
            await Task.Delay(10, ct);
            status[$"TRK{n:D3}"] = n % 3 == 0 ? "배송중" : "분류완료";
            Interlocked.Decrement(ref running);
        });

        Console.WriteLine($"조회 건수: {status.Count}");
        Console.WriteLine($"배송중: {status.Values.Count(v => v == "배송중")}");
        Console.WriteLine($"분류완료: {status.Values.Count(v => v == "분류완료")}");
        Console.WriteLine($"TRK003: {status["TRK003"]}");
        Console.WriteLine($"TRK004: {status["TRK004"]}");
        Console.WriteLine($"동시 실행 4 이하: {peak <= 4}");
    }
}

줄별 해설

경쟁 상태 재현

RaceDemo 의 스레드 본문은 값을 읽고(seen), SignalAndWait 로 상대가 읽기를 끝낼 때까지 기다린 뒤, seen + 1 을 쓴다. Barrier(2) 는 참가자 두 명이 모두 도착해야 통과시키므로 두 스레드는 항상 0을 읽고 나서 1을 쓴다. 그래서 결과는 실행마다 1이다. 스레드 순서에 기대는 재현이 아니라 순서를 고정한 재현이라는 점이 중요하다.

lock

GateCounter 는 딕셔너리와 잠금 객체를 함께 감춘다. Add 의 읽고 더하고 쓰는 세 단계가 하나의 lock 블록 안에 있어서 다른 스레드가 사이에 끼어들지 못한다. Snapshot 도 같은 잠금으로 읽는다. 읽는 쪽이 잠그지 않으면 딕셔너리가 바뀌는 도중을 볼 수 있다. 반환하는 것은 복사본 리스트라서 잠금 밖에서 안전하게 순회한다. 작업자 4명이 각각 25,000번씩 A와 B에 번갈아 더하므로 게이트별 합은 항상 50,000이다.

Interlocked

Increment 와 Add 는 건수와 합계를 원자적으로 올린다. 무게는 100 + (id % 50) * 10 이라 100에서 590까지 나오고, 1부터 1000까지 합하면 345,000이다. UpdateMax 는 현재 값을 읽고, 후보가 더 크면 CompareExchange 로 교체를 시도한다. 반환값 previous 가 읽었던 값과 같으면 교체가 성공한 것이다. 다르면 누군가 먼저 바꾼 것이므로 previous 를 새 현재 값으로 삼아 다시 비교한다.

동시 컬렉션

AddOrUpdate(키, 1, (_, old) => old + 1) 는 키가 없으면 1로 추가하고, 있으면 기존 값에 1을 더한다. 값 갱신 함수가 중복 호출될 수 있어도 결과가 맞도록 순수하게 작성했다. 출력은 Ordinal 정렬로 고정한다. 딕셔너리의 열거 순서에 기대면 안 된다.

Channel

용량 8의 유한 채널이다. 생산자 둘이 각각 500개를 넣고, 소비자 셋이 꺼내 지역별 건수와 무게를 집계한다. 소비자 안의 totalGram 은 여러 소비자가 함께 올리므로 Interlocked.Add 를 썼다. 생산자가 모두 끝난 뒤에 Complete() 를 호출하고, 그다음에 소비자를 기다린다. 이 순서가 바뀌어 소비자를 먼저 기다리면 프로그램이 끝나지 않는다. 무게는 500 + (id % 10) * 100 이므로 합은 950,000이다.

Parallel.ForEachAsync

본문 시작에서 running 을 올려 지금 실행 중인 본문 수를 세고, UpdateMax 로 최고치를 기록한다. 10밀리초 대기는 외부 조회를 흉내 낸 것이다. 3의 배수 번호는 "배송중", 나머지는 "분류완료"로 저장한다. 1부터 20 사이의 3의 배수는 6개이므로 나머지는 14개다. 출력에는 실제 최고 동시 수 대신 "4 이하인가"만 찍었다. 실제 최고치는 실행 환경에 따라 달라지기 때문이다.

실행 결과

$ dotnet run
[1] 경쟁 상태 재현
스캔 2회, 집계 1
[2] lock 으로 게이트별 집계
A: 50000
B: 50000
합계: 100000
[3] Interlocked 로 통계 집계
건수: 1000
총 무게(g): 345000
최대 무게(g): 590
[4] ConcurrentDictionary 로 지역별 집계
광주: 1000
대전: 1000
부산: 1000
서울: 1000
[5] Channel 생산자-소비자
처리 건수: 1000
총 무게(g): 950000
광주: 250
대전: 250
부산: 250
서울: 250
[6] Parallel.ForEachAsync 로 배송 조회
조회 건수: 20
배송중: 6
분류완료: 14
TRK003: 배송중
TRK004: 분류완료
동시 실행 4 이하: True

실무에서 자주 틀리는 것

lock 블록 안에서 await 하기

잠금 구간 안에서 비동기 호출을 하려다 컴파일 오류를 만난다.

// 틀린 코드: 컴파일되지 않는다(CS1996)
lock (_lock)
{
    var info = await _client.GetAsync(id);
    _cache[id] = info;
}

비동기 호출은 잠금 밖에서 끝내고, 결과를 저장하는 짧은 구간만 잠근다. 구간 전체를 보호해야 하면 SemaphoreSlim 을 쓴다.

// 고친 코드
var info = await _client.GetAsync(id);
lock (_lock)
{
    _cache[id] = info;
}

// 구간 전체를 보호해야 한다면
await _semaphore.WaitAsync();
try
{
    var info2 = await _client.GetAsync(id);
    _cache[id] = info2;
}
finally
{
    _semaphore.Release();
}

동시 딕셔너리에서 확인 후 쓰기

동시 컬렉션이니 안전하다고 믿고 확인과 쓰기를 따로 호출한다.

// 틀린 코드: 확인과 쓰기 사이에 다른 스레드가 끼어들 수 있다
if (!counts.ContainsKey(region))
{
    counts[region] = 0;
}
counts[region] = counts[region] + 1;

마지막 줄은 읽기-수정-쓰기라서 증가분이 사라진다. 한 번의 호출로 끝나는 메서드를 쓴다.

// 고친 코드
counts.AddOrUpdate(region, 1, (_, old) => old + 1);

Complete 를 호출하지 않기

생산자가 다 넣고 나면 소비자가 알아서 끝날 것으로 생각한다.

// 틀린 코드: 소비자 반복이 끝나지 않는다
await Task.WhenAll(producers);
await Task.WhenAll(consumers);

생산자가 예외로 끝나는 경우까지 생각해 try/finally 로 완료 신호를 보장한다.

// 고친 코드
try
{
    await Task.WhenAll(producers);
}
finally
{
    channel.Writer.Complete();
}
await Task.WhenAll(consumers);

ForEachAsync 본문에서 List 에 Add

결과를 모으려고 일반 컬렉션을 여러 본문이 함께 쓴다.

// 틀린 코드: List<T> 는 동시 쓰기에 안전하지 않다
var results = new List<string>();
await Parallel.ForEachAsync(ids, async (id, ct) =>
{
    results.Add(await LookupAsync(id, ct));
});

항목이 잘리거나 예외가 나거나 배열 내부가 깨질 수 있다. 순서가 필요 없으면 동시 컬렉션에 넣고, 순서가 필요하면 미리 크기를 정한 배열의 자기 자리에 쓴다.

// 고친 코드
var results = new ConcurrentBag<string>();
await Parallel.ForEachAsync(ids, async (id, ct) =>
{
    results.Add(await LookupAsync(id, ct));
});

// 순서가 필요하다면
var ordered = new string[ids.Length];
await Parallel.ForEachAsync(Enumerable.Range(0, ids.Length), async (i, ct) =>
{
    ordered[i] = await LookupAsync(ids[i], ct);
});

한눈에 보기

이 장의 API와 주의점 요약

주제	핵심 API	판단 기준	주의
경쟁 상태	읽기-수정-쓰기	공유 값을 여러 스레드가 갱신하는가	증가분 소실은 실행마다 달라 재현이 어려움
구간 보호	lock, Lock	여러 값이 함께 맞아야 하는가	구간 안에서 await 불가, 구간은 짧게
변수 하나	Interlocked	합계, 건수, 최댓값인가	변수 둘의 일관성은 보장하지 못함
동시 컬렉션	AddOrUpdate, TryAdd	개별 연산이 원자적이면 충분한가	확인 후 쓰기를 나누지 말 것
작업 전달	Channel	접수와 처리 속도가 다른가	유한 채널 사용, Complete 호출
비동기 병렬	Parallel.ForEachAsync	기다림이 대부분인 작업인가	순서 미보장, 본문의 공유 상태 보호

도구별 세부 동작은 공식 문서의 동기화 기본 요소 개요와 채널 문서에서 확인할 수 있다.

연습 문제

1. 작업자 4개가 각각 100,000번씩 _total 을 1씩 올리는 코드를 Interlocked 로 작성하고, 출력될 최종 값을 쓰시오.
2. 무게의 최솟값을 기록하는 UpdateMin(ref int target, int candidate) 를 CompareExchange 로 작성하시오. 초깃값은 int.MaxValue 로 가정한다.
3. 예제의 채널 용량을 1로 줄여도 결과가 같은 이유와, 줄였을 때 달라지는 점을 설명하시오. 또 Complete() 호출을 지우면 어떤 증상이 나타나는지 쓰시오.
4. 배송 조회 결과를 항목 번호 순서대로 출력해야 한다. Parallel.ForEachAsync 에서 결과를 어떤 방식으로 모으면 되는지 코드로 쓰시오.

정답과 해설

1. int total = 0;
   Parallel.For(0, 4, _ =>
   {
       for (int n = 0; n < 100_000; n++)
       {
           Interlocked.Increment(ref total);
       }
   });
   Console.WriteLine(total);
   

   출력은 400000이다. 람다가 캡처한 지역 변수에도 ref 를 쓸 수 있다. total++ 로 바꾸면 값이 400000보다 작아질 수 있고 실행마다 달라진다.

2. static void UpdateMin(ref int target, int candidate)
   {
       int current = Volatile.Read(ref target);
       while (candidate < current)
       {
           int previous = Interlocked.CompareExchange(ref target, candidate, current);
           if (previous == current)
           {
               return;
           }
           current = previous;
       }
   }
   

   부등호 방향만 UpdateMax 와 다르다. 교체가 실패하면 다른 스레드가 바꾼 값으로 다시 비교하는 것이 핵심이다.

3. 생산자는 가득 찬 채널에서 기다릴 뿐 항목을 잃지 않고, 소비자는 들어온 항목을 모두 꺼내므로 집계 결과는 같다. 달라지는 것은 속도와 메모리다. 용량이 작을수록 생산자가 자주 멈추고 채널에 쌓이는 항목이 적다. Complete() 를 지우면 소비자가 다음 항목을 계속 기다리므로 await Task.WhenAll(consumers) 에서 프로그램이 끝나지 않는다.

4. string[] ids = ["TRK001", "TRK002", "TRK003"];
   var results = new string[ids.Length];
   await Parallel.ForEachAsync(Enumerable.Range(0, ids.Length), async (i, ct) =>
   {
       await Task.Delay(10, ct);
       results[i] = $"{ids[i]} 조회 완료";
   });
   foreach (var line in results)
   {
       Console.WriteLine(line);
   }
   

   본문마다 서로 다른 인덱스 자리에만 쓰므로 여러 본문이 동시에 실행돼도 같은 칸을 건드리지 않는다. 완료 순서와 상관없이 출력 순서는 항목 번호 순이다. 자리를 나눌 수 없다면 ConcurrentBag 에 모은 뒤 정렬하면 된다.