Rust 채널: 스레드 간 메시지 전달

동시성 페이지에서는 스레드들이 Arc<Mutex<T>>로 데이터를 잠가서 공유했어요. 그 방법도 잘 동작하지만, 더 간단할 때가 많은 다른 방법이 있어요. 데이터를 공유하는 대신, 스레드끼리 메시지를 보내는 거예요.

채널(channel)은 양쪽 끝이 있는 파이프라고 생각하면 돼요. 한 스레드가 송신자(sender) 쪽에 값을 넣으면, 다른 스레드가 수신자(receiver) 쪽에서 넣은 순서대로 값을 꺼내요. 아무것도 공유하지 않아요. 값 하나하나가 한 스레드에서 다른 스레드로 이동할 뿐이에요. Go 커뮤니티에는 Rust도 빌려 쓰는 격언이 있어요. “메모리를 공유해서 통신하지 말고, 통신해서 메모리를 공유하라.”

첫 번째 채널

표준 라이브러리의 채널은 std::sync::mpsc에 있어요. 이름은 multiple producer, single consumer(여러 생산자, 단일 소비자)라는 뜻이에요. 송신자는 여럿, 수신자는 하나예요.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("작업 스레드가 보내는 인사");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main이 받은 값: {received}");
}

실행 결과:

main이 받은 값: 작업 스레드가 보내는 인사

▶ 직접 실행해 보기

한 단계씩 볼게요.

• mpsc::channel()은 Sender와 Receiver 한 쌍을 돌려줘요. 관례적으로 tx(transmit, 송신)와 rx(receive, 수신)라고 불러요.
• move는 tx를 새 스레드에게 넘겨요. 동시성 페이지에서 데이터를 스레드로 옮겼던 것과 같은 방식이에요.
• tx.send(value)는 채널에 값을 넣어요. 수신자가 이미 사라졌으면 실패하기 때문에 Result를 반환해요.
• rx.recv()는 값이 도착할 때까지 기다려요. 모든 송신자가 사라지고 채널이 비어 있을 때만 Err를 반환해요.

join()이 필요 없었다는 점도 눈여겨보세요. recv()가 이미 메시지를 기다리기 때문에 main이 너무 일찍 끝날 수 없어요.

여러 값 받기

Receiver는 반복자처럼 쓸 수 있어요. for 반복문으로 돌면 알아서 recv()를 계속 호출해 주고, 모든 송신자가 버려지면(drop) 반복문이 저절로 끝나요.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i * 10).unwrap();
        }
        // 스레드가 끝날 때 여기서 tx가 버려져요
    });

    for value in rx {
        println!("받음: {value}");
    }
    println!("채널이 닫혔어요. 모두 끝!");
}

실행 결과:

받음: 10
받음: 20
받음: 30
채널이 닫혔어요. 모두 끝!

▶ 직접 실행해 보기

여기서는 송신자가 하나뿐이라서, 값은 보낸 순서 그대로 도착해요. “송신자가 모두 사라지면 반복문이 멈춘다”는 규칙은 중요해요. 다음 절에서도 꼭 기억해 두세요.

여러 생산자: 송신자 복제하기

여러 스레드가 보낼 수 있게 하려면 tx.clone()으로 각 스레드에게 송신자 복사본을 하나씩 줘요. 모든 복사본은 같은 채널로 이어져요.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("{id}번 작업자 보고합니다")).unwrap();
        });
    }
    drop(tx); // 원본을 버려요. 안 그러면 아래 반복문이 끝나지 않아요

    let mut messages: Vec<String> = rx.iter().collect();
    messages.sort();
    for m in messages {
        println!("{m}");
    }
}

실행 결과:

1번 작업자 보고합니다
2번 작업자 보고합니다
3번 작업자 보고합니다

▶ 직접 실행해 보기

눈여겨볼 점이 두 가지 있어요.

• drop(tx)가 중요해요. 각 스레드는 끝날 때 자기 복제본을 버리지만, main은 아직 원본을 쥐고 있어요. 송신자가 하나라도 남아 있으면 수신자는 계속 기다리고, 프로그램은 영원히 멈춰 버려요. 원본을 버리는 건 “main은 아무것도 보내지 않을게”라고 알리는 거예요.
• 스레드는 어떤 순서로든 끝날 수 있어요. 메시지가 2번, 1번, 3번 작업자 순으로 도착할 수도 있어요. 그래서 메시지를 모아서 정렬하면 출력이 항상 같아져요. sort()를 지우고 몇 번 실행해서 순서가 바뀌는 걸 직접 확인해 보세요.

영원히 기다리지 않기: try_recv와 recv_timeout

recv()는 무언가 도착할 때까지 멈춰서 기다려요. 기다리고 싶지 않을 때도 있죠. 그럴 때 쓸 수 있는 친척이 둘 있어요.

• try_recv()는 바로 돌아와요. 값, Empty(아직 아무것도 없음), Disconnected(송신자가 모두 사라짐) 중 하나를 돌려줘요.
• recv_timeout(duration)은 기다리긴 하지만 제한 시간까지만 기다려요. 그 안에 아무것도 안 오면 Timeout을 받아요.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel::<i32>();

    // 아직 보낸 게 없어요. try_recv는 바로 돌아와요.
    println!("{:?}", rx.try_recv());

    thread::spawn(move || tx.send(7).unwrap()).join().unwrap();

    // 값이 이미 기다리고 있어서, 기다리지 않고 바로 돌려줘요.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // 송신자는 사라졌고 채널은 비어 있어요.
    println!("{:?}", rx.try_recv());
}

실행 결과:

Err(Empty)
Ok(7)
Err(Disconnected)

▶ 직접 실행해 보기

값을 받기 전에 스레드를 join()해 두었기 때문에, 값이 이미 기다리고 있고 출력도 항상 같아요.

try_recv()는 다른 할 일도 있는 반복문에서 편리해요. 예를 들어 프레임마다 새 메시지가 있는지 한 번씩 확인하는 게임 루프 같은 곳이요. recv_timeout()은 “작업 스레드가 너무 오래 걸리면 포기하기” 같은 경우에 좋아요.

sync_channel로 크기 제한이 있는 채널 만들기

보통의 channel()에는 크기 제한이 없어요. 송신자가 수신자보다 훨씬 빠르면 메시지가 메모리에 계속 쌓여요. mpsc::sync_channel(n)은 메시지를 최대 n개까지만 담는 채널을 만들어요. 채널이 가득 차면 send()는 수신자가 하나를 꺼낼 때까지 기다려요. 이걸 배압(backpressure)이라고 해요. 느린 소비자가 빠른 생산자의 속도를 자동으로 늦추는 거예요.

use std::sync::mpsc;
use std::thread;

fn main() {
    // 한 번에 메시지 2개만 담을 수 있어요.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // 버퍼가 가득 차 있는 동안은 기다려요
        }
    });

    let total: i32 = rx.iter().sum();
    producer.join().unwrap();
    println!("합계 = {total}");
}

실행 결과:

합계 = 15

▶ 직접 실행해 보기

생산자는 main보다 숫자 두 개 이상 앞서 나갈 수 없어요. sync_channel(0)이면 버퍼가 아예 없어요. send()할 때마다 수신자가 값을 받을 준비가 될 때까지 기다려요. 택배를 직접 손에서 손으로 건네는 것과 비슷해요.

작은 작업자 풀

이제 모두 합쳐 볼게요. 작업자 풀(worker pool)은 공유 큐에서 작업을 가져가서 결과를 돌려보내는 몇 개의 스레드예요. 채널을 두 개 써요. 하나는 작업용, 하나는 결과용이에요.

한 가지 함정이 있어요. Receiver는 복제할 수 없어요(단일 소비자니까요). 그래서 작업자들은 작업 수신자를 Arc<Mutex<_>>에 넣어 공유하고, 차례로 돌아가며 꺼내요.

use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();

    // 수신자는 복제할 수 없어서, 작업자들이 Arc<Mutex<_>>로 하나를 공유해요.
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _ in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(thread::spawn(move || loop {
            let job = job_rx.lock().unwrap().recv();
            match job {
                Ok(n) => result_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // 더 이상 작업이 없어요
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // 작업자들에게 더 할 일이 없다고 알려요

    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    for w in workers {
        w.join().unwrap();
    }
    results.sort();
    for (n, square) in results {
        println!("{n}의 제곱은 {square}");
    }
}

실행 결과:

1의 제곱은 1
2의 제곱은 4
3의 제곱은 9
4의 제곱은 16
5의 제곱은 25
6의 제곱은 36

▶ 직접 실행해 보기

깔끔하게 종료되는 과정은 이래요.

1. drop(job_tx)가 작업 채널을 닫아요. 그러면 각 작업자의 recv()가 Err를 반환하고, 작업자는 반복문을 빠져나가요.
2. 작업자가 끝나면 그 작업자의 result_tx 복제본이 버려져요. 원본은 이미 버렸으니, 세 작업자가 모두 끝나면 결과 채널이 닫혀요.
3. 그러면 result_rx.iter()가 멈추고, 작업자들이 끝나는 순서는 제각각이니 결과를 정렬해요.

작은 부분이지만, job_rx.lock().unwrap().recv()를 let 문 하나에 쓴 건 일부러 그런 거예요. 잠금은 그 문이 끝날 때 풀리기 때문에, 이 작업자가 아직 숫자를 제곱하는 동안 다른 작업자들이 다음 작업을 가져갈 수 있어요.

채널과 Arc<Mutex<T>> 비교

둘 다 안전해요. 다만 어울리는 문제의 모양이 달라요.

채널을 쓰면 좋을 때	Arc<Mutex<T>>를 쓰면 좋을 때
작업이 한 방향으로 흘러요. 생산자가 만들고, 소비자가 처리해요	여러 스레드가 같은 값을 읽고 고쳐야 해요(카운터, 캐시)
작업, 이벤트, 로그 줄의 큐가 필요해요	데이터가 커서 복사하거나 옮기면 낭비예요
각 값의 주인은 한 번에 하나예요	스레드가 업데이트의 흐름이 아니라 현재 상태를 봐야 해요

쓸 만한 경험칙이 있어요. 프로그램을 “이 스레드들이 저 스레드에게 무언가를 알려 준다“라고 설명할 수 있으면 채널을 쓰세요. “이 스레드들이 모두 이것을 고친다“라면 Mutex를 쓰세요. 실제 프로그램은 우리 작업자 풀처럼 둘 다 쓰는 경우가 많아요.

자주 하는 실수

보낸 값을 다시 쓰기. send()는 함수에 값을 넘길 때처럼 소유권을 가져가요. 보내고 나면 그 값은 수신자의 것이에요.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("안녕");
        tx.send(message).unwrap();
        println!("보냄: {message}");
    });

    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `message`
  --> src/main.rs:10:24
   |
 8 |         let message = String::from("안녕");
   |             ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(message).unwrap();
   |                 ------- value moved here
10 |         println!("보냄: {message}");
   |                          ^^^^^^^ value borrowed here after move

컴파일러는 “이동된 값 message를 빌려 쓰고 있다”고 알려 줘요. Rust가 여러분을 지켜 주는 거예요. 송신 쪽이 여전히 message를 바꿀 수 있다면 두 스레드가 같은 데이터를 건드리게 되니까요. 보내기 전에 출력하거나, .clone()을 보내세요.

스레드를 만드는 반복문 안으로 송신자를 옮기기. 첫 번째 스레드가 tx를 가져가 버려서, 두 번째 스레드에게는 남는 게 없어요.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        thread::spawn(move || {
            tx.send(id).unwrap();
        });
    }

    for value in rx {
        println!("{value}");
    }
}

error[E0382]: use of moved value: `tx`
  --> src/main.rs:8:23
   |
 5 |     let (tx, rx) = mpsc::channel();
   |          -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
...
 8 |         thread::spawn(move || {
   |                       ^^^^^^^ value moved into closure here, in previous iteration of loop

에러 메시지는 “반복문의 이전 회차에서 tx가 이미 클로저로 이동했다”는 뜻이에요. 해결 방법은 여러 생산자 예제처럼 반복문 맨 위에 let tx = tx.clone();을 쓰는 거예요.

끝나지 않는 프로그램. 이건 컴파일 에러가 나지 않아서 더 교묘해요. 어떤 송신자가 아직 살아 있는 동안(대개 main에 있는 원본 tx) 수신자를 반복문으로 돌면(for x in rx), 반복문은 영원히 기다려요. 프로그램이 “그냥 멈춰 있다”면, 모든 송신자가 버려지는지 확인해 보세요.

다음 단계

std::sync::mpsc로 많은 걸 할 수 있지만, 자주 쓰이는 대안 두 가지도 만나게 될 거예요.

• crossbeam-channel (영어)은 다중 소비자 채널(복제할 수 있는 수신자)과, 여러 채널을 한꺼번에 기다리는 select! 매크로를 제공해요.
• tokio::sync::mpsc 채널은 비동기 코드용이에요. async/.await를 쓴다면(async/await 참고) 표준 라이브러리 채널 대신 이걸 쓰세요. recv().await가 스레드 전체를 멈추지 않거든요.

정리

• mpsc::channel()은 (Sender, Receiver) 쌍을 줘요. send()는 값을 넣고, recv()는 다음 값을 기다려요.
• 생산자를 여럿 두려면 송신자를 복제하세요. 수신자는 언제나 하나뿐이에요.
• 수신자를 도는 반복문은 모든 송신자가 버려져야 끝나요. 그러니 원본을 drop()하는 걸 잊지 마세요.
• try_recv()와 recv_timeout()을 쓰면 영원히 기다리지 않고 확인할 수 있어요. sync_channel(n)은 큐를 메시지 n개로 제한해요.
• 여러 스레드에서 온 메시지는 어떤 순서로든 도착해요. 출력이 일정해야 한다면 정렬하거나 합산하세요.
• 데이터가 스레드에서 스레드로 흘러가면 채널을, 스레드들이 값 하나를 공유하며 고친다면 Arc<Mutex<T>>를 쓰세요.

이해도 점검

이해도 점검

1. "mpsc"는 무엇의 줄임말일까요?

Many processes, shared cache(여러 프로세스, 공유 캐시) Message passing, safe concurrency(메시지 전달, 안전한 동시성) Multiple producer, single consumer(여러 생산자, 단일 소비자) Mutex protected shared channel(뮤텍스로 보호되는 공유 채널)

2. 세 스레드가 같은 채널로 보낼 수 있게 하려면 어떻게 할까요?

각 스레드에게 tx.clone()을 줘요 수신자를 Rc로 감싸요 mpsc::channel()을 세 번 호출해서 합쳐요 불가능해요. 채널에는 송신자가 하나뿐이에요

3. for value in rx 반복문이 끝나지 않아요. 가장 그럴듯한 원인은 무엇일까요?

채널이 가득 찼어요 어떤 송신자(대개 main의 원본 tx)가 버려지지 않았어요 recv()를 너무 일찍 호출했어요 값이 Copy가 아니었어요

4. sync_channel(2)에 이미 메시지가 2개 들어 있을 때 send()는 어떻게 동작할까요?

가장 오래된 메시지를 버려요 바로 에러를 반환해요 패닉해요 수신자가 메시지를 하나 꺼낼 때까지 기다려요

점수: 0 / 4

더 알아보기

• 메시지 패싱을 사용하여 스레드 간 데이터 전송하기 — The Rust Programming Language 한국어판: 채널을 소개하는 공식 문서로, 예제가 더 많아요.
• std::sync::mpsc — 표준 라이브러리 문서 (영어): Sender, SyncSender, Receiver의 모든 메서드가 있어요.
• Channels — Rust by Example (영어): 생산자 스레드 여러 개를 쓰는 짧은 예제를 실행해 볼 수 있어요.
• 싱글스레드 서버를 멀티스레드 서버로 바꾸기 — The Rust Programming Language 한국어판: 채널과 Arc<Mutex<Receiver>>로 완전한 스레드 풀을 한 단계씩 만들어요.
• tokio::sync::mpsc — docs.rs (영어): Tokio에서 쓰는 비동기 채널과, 표준 라이브러리 채널과의 차이를 설명해요.