Go 의 동시성 (concurrency) 더 들여다보기

package main

import (
	"fmt"
	"sync"
)

func main() {
	jobs := make(chan int, 5)
	results := make(chan int, 5)
	var wg sync.WaitGroup

	// 3개의 워커 고루틴 생성
	for w := 1; w <= 3; w++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for j := range jobs {
				fmt.Printf("워커 %d, 작업 %d 시작\n", id, j)
				results <- j * j // 결과를 results 채널에 보냄
			}
		}(w)
	}

	// 5개의 작업을 jobs 채널에 보냄
	for j := 1; j <= 5; j++ {
		jobs <- j
	}
	close(jobs) // 작업 전송이 끝났음을 알림

	wg.Wait() // 모든 워커가 끝날 때까지 대기
	close(results)

	// 결과 수집
	for r := range results {
		fmt.Printf("결과: %d\n", r)
	}
}

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class ConcurrencyTest {
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        List<Future<Integer>> futures = new ArrayList<>();
        List<Integer> inputs = List.of(1, 2, 3, 4, 5);

        for (int input : inputs) {
            Callable<Integer> task = () -> {
                System.out.println("워커 " + Thread.currentThread().getName() + ", 작업 " + input + " 시작");
                return input * input;
            };
            futures.add(executor.submit(task));
        }

        for (Future<Integer> future : futures) {
            System.out.println("결과: " + future.get()); // 결과가 준비될 때까지 블로킹
        }

        executor.shutdown();
    }
}

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel

suspend fun main() = coroutineScope {
    val jobs = Channel<Int>()
    val results = Channel<Int>()

    // 5개의 작업을 비동기적으로 보냄
    launch {
        for (i in 1..5) {
            jobs.send(i)
        }
        jobs.close()
    }

    // 3개의 워커 코루틴
    repeat(3) { id ->
        launch {
            for (j in jobs) {
                println("워커 $id, 작업 $j 시작")
                results.send(j * j)
            }
        }
    }
    
    // 결과 5개를 수집 후 종료
    repeat(5) {
        println("결과: ${results.receive()}")
    }
    coroutineContext.cancelChildren() // 모든 자식 코루틴 정리
}

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

if (isMainThread) {
    const inputs = [1, 2, 3, 4, 5];
    const numWorkers = 3;
    let pending = inputs.length;

    console.log(`메인 스레드: ${inputs.length}개의 작업을 ${numWorkers}개의 워커에 분배`);

    for (let i = 0; i < numWorkers; i++) {
        const worker = new Worker(__filename);
        worker.on('message', (result) => {
            console.log(`결과: ${result}`);
            if (--pending === 0) {
                console.log('모든 작업 완료');
            }
        });
        worker.on('error', (err) => console.error(err));
    }
} else {
    // 워커 스레드
    parentPort.on('message', (job) => {
        console.log(`워커, 작업 ${job} 시작`);
        parentPort.postMessage(job * job);
    });
}
// 실제 구현에서는 메인 스레드가 워커에 작업을 분배하는 로직이 필요.
// 이 예제는 worker_threads의 구조를 보여주기 위함.

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

// Rust의 표준 라이브러리인 mpsc(multiple producer, single consumer) 채널의 특성 반영
fn main() {
    // 작업(jobs)과 결과(results)를 위한 두 개의 채널 생성
    let (jobs_tx, jobs_rx) = mpsc::channel();
    let (results_tx, results_rx) = mpsc::channel();

    // 여러 워커 스레드가 'jobs_rx'를 공유하기 위해 Arc<Mutex<>>로 감싼다.
    let jobs_rx = Arc::new(Mutex::new(jobs_rx));
    let mut handles = vec![];

    // 3개의 워커 스레드를 생성
    for id in 0..3 {
        // 각 스레드가 채널의 소유권을 가질 수 있도록 복제
        let jobs_rx_clone = Arc::clone(&jobs_rx);
        let results_tx_clone = results_tx.clone();

        let handle = thread::spawn(move || {
            loop {
                // Mutex를 잠그고(lock) 채널에서 작업을 하나 수신(recv)한다.
                // Mutex는 한 번에 하나의 스레드만 jobs_rx에 접근하도록 보장한다.
                let job_result = jobs_rx_clone.lock().unwrap().recv();

                match job_result {
                    // 작업 수신에 성공한 경우
                    Ok(job) => {
                        println!("워커 {}, 작업 {} 시작", id, job);
                        let result = job * job; // 작업 수행 (제곱)
                        results_tx_clone.send(result).unwrap(); // 결과를 results 채널로 전송
                    }
                    // 채널이 닫혀 더 이상 받을 작업이 없는 경우
                    Err(_) => {
                        println!("워커 {}, 종료", id);
                        break; // 루프를 종료하여 스레드를 끝낸다.
                    }
                }
            }
        });
        handles.push(handle);
    }

    // 메인 스레드에서 더 이상 results_tx를 사용하지 않으므로 소유권을 포기한다.
    // 모든 워커 스레드의 results_tx_clone이 drop되어야 results_rx가 종료될 수 있다.
    drop(results_tx);

    // 5개의 작업을 jobs 채널에 보낸다.
    let inputs = vec![1, 2, 3, 4, 5];
    for job in inputs {
        jobs_tx.send(job).unwrap();
    }

    // 모든 작업을 보낸 후 jobs_tx의 소유권을 포기한다.
    // 송신자(tx)가 모두 사라지면 수신자(rx)는 에러(Err)를 반환받게 된다.
    // 이것이 워커 스레드에게 작업이 끝났음을 알리는 신호가 된다.
    drop(jobs_tx);

    // 결과 수집
    for received in results_rx {
        println!("결과: {}", received);
    }

    // 모든 워커 스레드가 작업을 마칠 때까지 대기
    for handle in handles {
        handle.join().unwrap();
    }
}