mpsc - Topics about IT

1. 동시성(Concurrency)이란?

동시성(Concurrency)은 프로그램이 여러 작업을 “동시에” 처리하는 능력을 말합니다. 꼭 물리적으로 동시에 실행되는 것(병렬성)이 아니라, 여러 작업을 논리적으로 동시에 실행하는 것입니다. 즉, 동시성은 스레드보다 더 큰 개념이며, 스레드는 그 구현 수단 중 하나입니다.

Rust에서 동시성은 다음과 같은 방식으로 구현할 수 있습니다:

스레드(Thread)
메시지 전달(Channel)
공유 메모리(Arc<Mutex<T>>)
비동기 프로그래밍(async/await)

2. Rust의 동시성 철학

Rust의 동시성은 다음 원칙을 따릅니다:

안전성 (Safety): 동시성 버그(데이터 경쟁, 데드락 등)를 컴파일 타임에 방지
명시성 (Explicitness): 공유 자원, 락, 메시지 전달 등이 명확히 드러남
추론 가능성: 컴파일러가 Send, Sync 트레잇으로 스레드 간 이동 가능 여부를 추론

C++/Java는 런타임 오류가 나야 알 수 있는 동시성 문제를, Rust는 컴파일러가 사전에 차단합니다.

3. 스레드를 이용한 동시성

use std::thread;

fn main() {
    let handle = thread::spawn(|| {
        println!("스레드에서 실행 중");
    });

    println!("메인 스레드 실행 중");
    handle.join().unwrap();
}

thread:spawn을 통해 새로운 스레드를 생성하고, join()을 통해 스레드 종료를 기다립니다.
이 방식은 동시에 여러 작업을 실행하려는 목적, 즉 동시성을 달성하는 가장 기본적인 방식입니다.

use std::thread;

Rust 표준 라이브러리의 thread 모듈을 가져옵니다(import).
이 모듈은 스레드 생성 및 제어에 관련된 기능을 제공합니다.

let handle = thread::spawn(|| { … });

thread::spawn은 새로운 스레드를 생성하고 그 안에서 클로저(익명 함수)를 실행합니다.
이 예제에서 생성된 스레드는 println!(“스레드에서 실행 중”);을 실행합니다.
spawn은 JoinHandle이라는 객체를 반환하며, 이를 handle 변수에 저장합니다.
이 handle은 이후 스레드가 언제 종료됐는지 기다릴 수 있게(join) 도와줍니다.

println!(“메인 스레드 실행 중”);

메인 스레드는 새로 만든 스레드와 동시에 실행됩니다.
따라서 이 줄과 스레드에서 실행 중 메시지 중 어느 것이 먼저 출력될지는 실행 시마다 달라질 수 있습니다.
이는 Rust의 동시성(Concurrency) 특징입니다.

handle.join().unwrap();

join()은 새로 생성한 스레드가 끝날 때까지 기다리는 함수입니다.
스레드 실행이 끝나면 Result<T, E>를 반환하고, 성공하면 Ok(()), 실패하면 Err(e)가 반환됩니다.
여기서는 unwrap()을 사용해에러가 발생하면 패닉(panic)하도록 합니다.

4. 메시지 전달로 동시성 구현 (채널)

Rust는 공유 메모리 대신 메시지 전달(Message Passing) 모델을 권장합니다. std::sync::mpsc 모듈을 통해 채널을 만들 수 있습니다.

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

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

    thread::spawn(move || {
        tx.send("데이터").unwrap();
    });

    let received = rx.recv().unwrap();
    println!("수신: {}", received);
}

위 코드는 멀티 스레드 환경에서 채널(channel)을 이용한 스레드 간 통신을 보여줍니다.

use std::sync::mpsc;

mpsc는 “multiple producer, single consumer”의 약자로, 여러 개의 생산자(보내는 쪽)가 하나의 소비자(받는 쪽)를 지원하는, 다시 말해 여러 생산자(스레드)가 하나의 소비자(스레드)에게 데이터를 보낼 수 있는 채널(channel) 시스템입니다.

let (tx, rx) = mpsc::channel();

channel() 함수는 전송자(Sender) 와 수신자(Receiver) 한 쌍을 반환합니다.
- tx: 데이터 전송에 사용 (transmitter)
- rx: 데이터 수신에 사용 (receiver)

thread::spawn(move || { … })

새로운 스레드를 생성합니다.
move 키워드는 클로저 내부로 tx를 소유권 이동(move)시키기 위해 사용됩니다.
tx를 다른 스레드에서 사용할 수 있게 하려면 반드시 move가 필요합니다.

tx.send(“데이터”).unwrap();

생성된 스레드에서 “데이터”라는 문자열을 메인 스레드로 전송합니다.
전송 결과가 Result이기 때문에, .unwrap()을 호출하여 에러가 나면 패닉하게 합니다.

let received = rx.recv().unwrap();

메인 스레드에서 recv()로 데이터가 도착할 때까지 블로킹(기다림)합니다.
다른 스레드에서 “데이터”를 보냈기 때문에, 수신이 성공하면 received에 저장됩니다.

println!(“수신: {}”, received);

받은 데이터를 출력합니다.

출력 결과는 “수신: 데이터”입니다.

공유된 메모리에 락을 걸 필요 없이, 스레드 간 메시지를 전달함으로써 동시성을 구현합니다.

이 방식은 Go 언어의 고루틴 + 채널과 비슷하며, 병렬성이 아닌 동시성 컨트롤에 적합합니다.

5. Arc<Mutex<T>>를 통한 공유 메모리 동시성

다수의 스레드가 데이터를 공유해야 한다면, Arc로 참조를 공유하고 Mutex로 락을 걸어야 합니다.

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

fn main() {
    let data = Arc::new(Mutex::new(0));

    let mut handles = vec![];
    for _ in 0..5 {
        let data = Arc::clone(&data);
        handles.push(thread::spawn(move || {
            let mut num = data.lock().unwrap();
            *num += 1;
        }));
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("최종 값: {}", *data.lock().unwrap());
}

위 코드는 여러 스레드가 안전하게 하나의 데이터를 공유하고 수정하는 예제입니다. 여기서는 Mutex와 Arc를 조합하여 스레드 간 공유 데이터의 동기화를 구현하고 있습니다.

가. 주요 개념 정리

(1) Mutex<T>: 상호 배제

여러 스레드가 동시에 데이터에 접근하지 못하도록 막아주는 락(lock).
.lock().unwrap()을 호출하면 락을 얻고, 반환된 객체(MutexGuard)를 통해 데이터를 읽고 쓸 수 있음.
락을 가진 객체는 스코프를 벗어나면 자동으로 해제됨.

(2) Arc<T>: 원자적 참조 카운트

여러 스레드에서 읽기 전용 또는 Mutex와 결합한 공유 사용이 가능.
Rc는 단일 스레드용이고, Arc는 멀티스레드에서도 안전하게 참조를 공유할 수 있도록 설계됨.
Arc::clone(&data)는 참조 카운트를 늘릴 뿐, 실제 데이터를 복사하지 않음.

나. 코드 분석

let data = Arc::new(Mutex::new(0));

0이라는 값을 Mutex로 감싸고, 이를 다시 Arc로 감쌈.
이렇게 하면 여러 스레드가 이 데이터를 공유하면서도 동시에 수정하지 못하도록 보장할 수 있음.

let mut handles = vec![];

for _ in 0..5 {
    let data = Arc::clone(&data);
    handles.push(thread::spawn(move || {
        let mut num = data.lock().unwrap();
        *num += 1;
    }));
}

5개의 스레드를 생성.
각 스레드는 Arc::clone(&data)를 통해 공유 데이터를 가리키는 참조를 가짐.
data.lock().unwrap()을 통해 락을 획득하고 *num += 1로 값을 증가시킴.
unwrap()은 락 획득 실패(예: 패닉) 시 프로그램을 종료시키는 간단한 오류 처리 방식.

for handle in handles {
    handle.join().unwrap();
}

각 스레드가 종료될 때까지 대기 (join()).
이렇게 해야 모든 스레드가 작업을 마치고 최종 값을 출력함.

println!("최종 값: {}", *data.lock().unwrap());

메인 스레드에서 최종 값을 읽음.
이때도 lock()을 사용하여 락을 획득한 뒤 출력.

다. 결과

각 스레드는 0부터시작한 값을 1씩 증가시킴.
5개의 스레드가 있으므로 최종 값은 5가 됨.

6. 비동기(async/await) 기반 동시성

Rust는 스레드 외에도 async fn, Future, await 등을 이용한 비동기 동시성 모델도 지원합니다. 이 모델은 싱글 스레드에서도 여러 작업을 동시에 처리하는 방식으로, 특히 IO 작업에 적합합니다.

use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let f1 = async {
        sleep(Duration::from_secs(1)).await;
        println!("1초 작업 완료");
    };

    let f2 = async {
        println!("즉시 실행");
    };

    tokio::join!(f1, f2);
}

use tokio::time::{sleep, Duration};

tokio 라이브러리의 time 모듈에서 sleep함수와 Duration 타입을 불러옵니다(import).

가. 코드 분석

#[tokio::main]

이 어트리뷰트는 비동기 함수 main을 실행할 수 있도록 tokio 런타임을 자동으로 생성해줍니다.
main() 함수가 async fn이므로 일반적으로는 실행할 수 없는데, tokio가 런타임을 초기화하고 실행을 담당합니다.

let f1 = async { … }

비동기 블록입니다.
내부에서sleep(Duration::from_secs(1)).await를 사용하여 1초간 대기한 후, “1초 작업 완료”를 출력합니다.
sleep은 스레드를 멈추는 것이 아니라, 해당 Future를 잠시 중단(suspend) 시켜 다른 작업을 실행할 수 있게 합니다.

let f2 = async { … }

이 비동기 블록은 바로 “즉시 실행”을 출력합니다.
await나 지연 없이 즉시 완료되는 작업입니다.

tokio::join!(f1, f2);

이 매크로는 f1과 f2를 동시에 실행시킵니다.
두 작업이 모두 끝날 때까지 기다립니다.
즉, f2는 바로 실행되고 “즉시 실행”이 출력됩니다.
f1은 1초 후 “1초 작업 완료”를 출력합니다.

나. 실행 결과

실제로 프로그램이 출력하는 순서는 다음과 같습니다:

즉시 실행
1초 작업 완료

두 작업은 병렬이 아닌 동시에 스케줄되는 병행(concurrent) 실행입니다.

비동기 동시성은 Tokio 같은런타임이 필요합니다. 이는 경량 스레드처럼 동작하여 수천 개의 작업을 동시 처리할 수 있습니다.

7. 정리

구분	특징	사용 예
스레드	OS 레벨 병렬 실행	CPU 작업 처리
채널	메시지 기반 동시성	안전한 데이터 전달
Arc<Mutex<T>>	메모리 공유 및 제어	공유 자원 카운터 등
async/await	경량 동시성, 비차단	네트워크, 파일 IO

8. 마무리

Rust의 동시성은 단순히 스레드를 사용하는 것에 그치지 않고, 메시지 전달 모델, 공유 메모리 모델, 비동기 실행 모델 등 다양한 방식으로 구현될 수 있습니다. Rust는 이러한 모델을 안전하고 명시적으로 구현할 수 있도록 돕는 언어이며, 컴파일 타임에 동시성 버그를 방지하는 강력한 도구입니다.

Rust에서 동시성은 단지 기술이 아니라, 신뢰성과 안전성을 보장하는 언어적 철학입니다.

[태그:] mpsc

동시성(Concurrency)과 스레드(Thread)