# Python: threading モジュール
import threading
import time
 
def worker(name, delay):
    print(f"Thread {name} started")
    time.sleep(delay)  # I/O を模擬
    print(f"Thread {name} finished after {delay}s")
 
# スレッドの作成と実行
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(f"Worker-{i}", i * 0.5))
    threads.append(t)
    t.start()
 
for t in threads:
    t.join()  # 全スレッドの完了を待つ
 
print("All threads completed")
 
# 注意: Python の GIL（Global Interpreter Lock）
# → CPython では1度に1スレッドしか Python コードを実行できない
# → CPU密集型処理には multiprocessing を使う
# → I/O密集型処理には threading が有効
 
# デーモンスレッド
daemon_thread = threading.Thread(target=worker, args=("Daemon", 10), daemon=True)
daemon_thread.start()
# メインスレッドが終了するとデーモンスレッドも自動終了
 
# スレッドプール（concurrent.futures）
from concurrent.futures import ThreadPoolExecutor, as_completed
 
def fetch_url(url):
    """URL からデータを取得する（模擬）"""
    time.sleep(1)  # ネットワーク遅延を模擬
    return f"Data from {url}"
 
urls = [
    "https://api.example.com/users",
    "https://api.example.com/posts",
    "https://api.example.com/comments",
    "https://api.example.com/albums",
    "https://api.example.com/photos",
]
 
# 最大3スレッドのプールで並行実行
with ThreadPoolExecutor(max_workers=3) as executor:
    # submit: 個別にタスクを投入
    futures = {executor.submit(fetch_url, url): url for url in urls}
 
    for future in as_completed(futures):
        url = futures[future]
        try:
            data = future.result()
            print(f"{url}: {data}")
        except Exception as e:
            print(f"{url} generated an exception: {e}")
 
# map: 全タスクを一括投入
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(fetch_url, urls))
    for url, data in zip(urls, results):
        print(f"{url}: {data}")

# Python: multiprocessing（GIL を回避する CPU 密集型処理向け）
from multiprocessing import Process, Pool, Queue, Value, Array
import os
 
def cpu_intensive_task(n):
    """CPU密集型の計算"""
    result = sum(i * i for i in range(n))
    return result
 
# Process の直接使用
def worker_process(name):
    pid = os.getpid()
    print(f"Process {name} (PID: {pid}) started")
    result = cpu_intensive_task(10_000_000)
    print(f"Process {name} result: {result}")
 
processes = []
for i in range(4):
    p = Process(target=worker_process, args=(f"P-{i}",))
    processes.append(p)
    p.start()
 
for p in processes:
    p.join()
 
# プロセスプール
with Pool(processes=4) as pool:
    # map: 同期的に全タスクを実行
    results = pool.map(cpu_intensive_task, [1_000_000] * 8)
    print(f"Results: {results}")
 
    # apply_async: 非同期にタスクを投入
    async_results = [
        pool.apply_async(cpu_intensive_task, (1_000_000,))
        for _ in range(8)
    ]
    results = [r.get(timeout=30) for r in async_results]
    print(f"Async results: {results}")
 
# プロセス間通信: Queue
def producer(queue):
    for i in range(10):
        queue.put(f"item-{i}")
    queue.put(None)  # 終了シグナル
 
def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"Consumed: {item}")
 
q = Queue()
prod = Process(target=producer, args=(q,))
cons = Process(target=consumer, args=(q,))
prod.start()
cons.start()
prod.join()
cons.join()
 
# 共有メモリ: Value, Array
counter = Value('i', 0)  # 'i' = int型
shared_array = Array('d', [0.0] * 10)  # 'd' = double型
 
def increment(counter, lock):
    for _ in range(100000):
        with lock:
            counter.value += 1
 
lock = multiprocessing.Lock()
ps = [Process(target=increment, args=(counter, lock)) for _ in range(4)]
for p in ps:
    p.start()
for p in ps:
    p.join()
print(f"Counter: {counter.value}")  # → 400000

// Rust: std::thread（OSスレッド）
use std::thread;
use std::time::Duration;
 
fn main() {
    // スレッドの生成
    let handle = thread::spawn(|| {
        println!("Hello from thread! ID: {:?}", thread::current().id());
        thread::sleep(Duration::from_millis(100));
        42  // スレッドの戻り値
    });
 
    let result = handle.join().unwrap();  // → 42
    println!("Thread returned: {}", result);
 
    // 複数スレッドの生成
    let handles: Vec<_> = (0..5)
        .map(|i| {
            thread::spawn(move || {
                println!("Thread {} started", i);
                thread::sleep(Duration::from_millis(100 * i as u64));
                println!("Thread {} finished", i);
                i * i
            })
        })
        .collect();
 
    let results: Vec<_> = handles
        .into_iter()
        .map(|h| h.join().unwrap())
        .collect();
    println!("Results: {:?}", results);  // [0, 1, 4, 9, 16]
 
    // スレッドビルダー（名前とスタックサイズの設定）
    let builder = thread::Builder::new()
        .name("custom-thread".into())
        .stack_size(32 * 1024);  // 32KB スタック
 
    let handle = builder.spawn(|| {
        let name = thread::current().name().unwrap_or("unnamed").to_string();
        println!("Running in thread: {}", name);
    }).unwrap();
 
    handle.join().unwrap();
}
 
// データの共有（Arc + Mutex）
use std::sync::{Arc, Mutex};
 
fn shared_counter_example() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];
 
    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();
            *num += 1;
            // MutexGuard がドロップされるとロック解放
        });
        handles.push(handle);
    }
 
    for handle in handles {
        handle.join().unwrap();
    }
    println!("Result: {}", *counter.lock().unwrap());  // → 10
}
 
// RwLock（読み書きロック）
use std::sync::RwLock;
 
fn rwlock_example() {
    let config = Arc::new(RwLock::new(HashMap::new()));
 
    // 複数の読み取りスレッド
    let readers: Vec<_> = (0..5)
        .map(|i| {
            let config = Arc::clone(&config);
            thread::spawn(move || {
                let data = config.read().unwrap();
                println!("Reader {}: {:?}", i, *data);
            })
        })
        .collect();
 
    // 1つの書き込みスレッド
    {
        let config = Arc::clone(&config);
        thread::spawn(move || {
            let mut data = config.write().unwrap();
            data.insert("key", "value");
            println!("Writer: inserted data");
        })
        .join()
        .unwrap();
    }
 
    for r in readers {
        r.join().unwrap();
    }
}
 
// Atomic 操作（ロックフリー）
use std::sync::atomic::{AtomicI64, Ordering};
 
fn atomic_example() {
    let counter = Arc::new(AtomicI64::new(0));
    let handles: Vec<_> = (0..10)
        .map(|_| {
            let counter = Arc::clone(&counter);
            thread::spawn(move || {
                for _ in 0..1000 {
                    counter.fetch_add(1, Ordering::Relaxed);
                }
            })
        })
        .collect();
 
    for h in handles {
        h.join().unwrap();
    }
    println!("Atomic counter: {}", counter.load(Ordering::Relaxed)); // → 10000
}

package main
 
import (
    "fmt"
    "runtime"
    "sync"
    "sync/atomic"
    "time"
)
 
// Go: goroutine（軽量スレッド）
func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Printf("Worker %d started (goroutine)\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %d done\n", id)
}
 
func main() {
    // 使用するCPUコア数を設定
    runtime.GOMAXPROCS(runtime.NumCPU())
 
    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go worker(i, &wg) // goroutine を起動（goキーワード）
    }
    wg.Wait() // 全goroutineの完了を待つ
 
    // goroutine は OSスレッドより遥かに軽量
    // → 数百万の goroutine を同時に実行可能
    // → Go ランタイムが M:N スケジューリング
    // → 1 goroutine のスタックは 2KB から開始（動的に成長）
 
    // 大量のgoroutineを生成する例
    var wg2 sync.WaitGroup
    count := int64(0)
    for i := 0; i < 100000; i++ {
        wg2.Add(1)
        go func() {
            defer wg2.Done()
            atomic.AddInt64(&count, 1)
        }()
    }
    wg2.Wait()
    fmt.Printf("Count: %d\n", count) // → 100000
}
 
// Mutex を使ったスレッドセーフなカウンター
type SafeCounter struct {
    mu sync.Mutex
    v  map[string]int
}
 
func (c *SafeCounter) Inc(key string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.v[key]++
}
 
func (c *SafeCounter) Value(key string) int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.v[key]
}
 
// RWMutex を使ったキャッシュ
type Cache struct {
    mu   sync.RWMutex
    data map[string]interface{}
}
 
func (c *Cache) Get(key string) (interface{}, bool) {
    c.mu.RLock()         // 読み取りロック（複数同時OK）
    defer c.mu.RUnlock()
    val, ok := c.data[key]
    return val, ok
}
 
func (c *Cache) Set(key string, value interface{}) {
    c.mu.Lock()          // 書き込みロック（排他）
    defer c.mu.Unlock()
    c.data[key] = value
}
 
// sync.Once: 一度だけ実行を保証
var (
    instance *Database
    once     sync.Once
)
 
type Database struct {
    // ...
}
 
func GetDB() *Database {
    once.Do(func() {
        instance = &Database{}
        fmt.Println("Database initialized")
    })
    return instance
}

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
 
// Java: Thread クラスと Runnable インターフェース
public class ThreadExample {
 
    // 方法1: Thread クラスを継承
    static class MyThread extends Thread {
        @Override
        public void run() {
            System.out.println("Thread: " + getName() + " running");
        }
    }
 
    // 方法2: Runnable インターフェースを実装
    static class MyRunnable implements Runnable {
        @Override
        public void run() {
            System.out.println("Runnable running in: " + Thread.currentThread().getName());
        }
    }
 
    public static void main(String[] args) throws Exception {
        // Thread クラス
        MyThread t1 = new MyThread();
        t1.start();
        t1.join();
 
        // Runnable（ラムダ式）
        Thread t2 = new Thread(() -> {
            System.out.println("Lambda thread running");
        });
        t2.start();
        t2.join();
 
        // ExecutorService（スレッドプール）
        ExecutorService executor = Executors.newFixedThreadPool(4);
 
        // submit: Future を返す
        Future<Integer> future = executor.submit(() -> {
            Thread.sleep(1000);
            return 42;
        });
        System.out.println("Result: " + future.get()); // → 42
 
        // invokeAll: 全タスクの完了を待つ
        var tasks = List.of(
            (Callable<String>) () -> { Thread.sleep(100); return "A"; },
            (Callable<String>) () -> { Thread.sleep(200); return "B"; },
            (Callable<String>) () -> { Thread.sleep(300); return "C"; }
        );
        var futures = executor.invokeAll(tasks);
        for (var f : futures) {
            System.out.println(f.get());
        }
 
        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);
 
        // Java 21+: Virtual Threads（Project Loom）
        // 軽量な仮想スレッド（goroutine に類似）
        try (var vExecutor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < 100000; i++) {
                final int id = i;
                vExecutor.submit(() -> {
                    // 各仮想スレッドで実行
                    Thread.sleep(Duration.ofMillis(100));
                    return "VThread-" + id;
                });
            }
        }
    }
}
 
// AtomicInteger: ロックフリーなカウンター
class AtomicCounter {
    private final AtomicInteger count = new AtomicInteger(0);
 
    public void increment() {
        count.incrementAndGet();
    }
 
    public int get() {
        return count.get();
    }
 
    // CAS（Compare-And-Swap）操作
    public boolean compareAndSet(int expected, int newValue) {
        return count.compareAndSet(expected, newValue);
    }
}

#include <thread>
#include <mutex>
#include <shared_mutex>
#include <atomic>
#include <future>
#include <vector>
#include <iostream>
 
// C++11 以降の std::thread
void basic_thread_example() {
    // スレッド生成
    std::thread t([] {
        std::cout << "Hello from thread " << std::this_thread::get_id() << std::endl;
    });
    t.join(); // 完了を待つ
 
    // detach: スレッドを切り離す（デーモンスレッド化）
    std::thread daemon([] {
        // バックグラウンド処理...
    });
    daemon.detach();
}
 
// Mutex による排他制御
std::mutex mtx;
int shared_counter = 0;
 
void increment_with_mutex() {
    for (int i = 0; i < 100000; i++) {
        std::lock_guard<std::mutex> lock(mtx); // RAII ロック
        shared_counter++;
    } // lock_guard のデストラクタでアンロック
}
 
// shared_mutex（読み書きロック、C++17）
std::shared_mutex rw_mutex;
std::map<std::string, std::string> cache;
 
std::string read_cache(const std::string& key) {
    std::shared_lock lock(rw_mutex); // 読み取りロック
    auto it = cache.find(key);
    return it != cache.end() ? it->second : "";
}
 
void write_cache(const std::string& key, const std::string& value) {
    std::unique_lock lock(rw_mutex); // 書き込みロック
    cache[key] = value;
}
 
// std::async と std::future
void async_example() {
    // 非同期タスクの実行
    auto future = std::async(std::launch::async, [] {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        return 42;
    });
 
    // 他の処理を行う...
    std::cout << "Waiting for result..." << std::endl;
 
    int result = future.get(); // 結果を取得（ブロッキング）
    std::cout << "Result: " << result << std::endl;
}
 
// atomic 操作
std::atomic<int> atomic_counter{0};
 
void atomic_increment() {
    for (int i = 0; i < 100000; i++) {
        atomic_counter.fetch_add(1, std::memory_order_relaxed);
    }
}

// JavaScript はシングルスレッドだが、並行処理の同期は必要
 
// Semaphore の実装
class Semaphore {
    private queue: Array<() => void> = [];
    private count: number;
 
    constructor(maxConcurrency: number) {
        this.count = maxConcurrency;
    }
 
    async acquire(): Promise<void> {
        if (this.count > 0) {
            this.count--;
            return;
        }
        return new Promise<void>((resolve) => {
            this.queue.push(resolve);
        });
    }
 
    release(): void {
        const next = this.queue.shift();
        if (next) {
            next();
        } else {
            this.count++;
        }
    }
}
 
// 使用例: API 呼び出しの並行数制限
async function fetchWithLimit(urls: string[], maxConcurrency: number) {
    const semaphore = new Semaphore(maxConcurrency);
    const results = await Promise.all(
        urls.map(async (url) => {
            await semaphore.acquire();
            try {
                const response = await fetch(url);
                return response.json();
            } finally {
                semaphore.release();
            }
        })
    );
    return results;
}
 
// Mutex の実装（async/await ベース）
class AsyncMutex {
    private locked = false;
    private queue: Array<() => void> = [];
 
    async lock(): Promise<() => void> {
        if (!this.locked) {
            this.locked = true;
            return () => this.unlock();
        }
        return new Promise<() => void>((resolve) => {
            this.queue.push(() => {
                resolve(() => this.unlock());
            });
        });
    }
 
    private unlock(): void {
        const next = this.queue.shift();
        if (next) {
            next();
        } else {
            this.locked = false;
        }
    }
}
 
// Web Workers（ブラウザでの並列処理）
// main.js
const worker = new Worker("worker.js");
worker.postMessage({ data: [1, 2, 3, 4, 5], operation: "sum" });
worker.onmessage = (event) => {
    console.log("Result from worker:", event.data);
};
 
// worker.js
// self.onmessage = (event) => {
//     const { data, operation } = event.data;
//     if (operation === "sum") {
//         const result = data.reduce((a, b) => a + b, 0);
//         self.postMessage(result);
//     }
// };
 
// SharedArrayBuffer と Atomics（共有メモリ）
const buffer = new SharedArrayBuffer(1024);
const view = new Int32Array(buffer);
 
// メインスレッド
Atomics.store(view, 0, 42);
Atomics.notify(view, 0);
 
// ワーカースレッド
// Atomics.wait(view, 0, 0); // 値が 0 でなくなるまで待機
// const value = Atomics.load(view, 0); // → 42

import threading
from collections import deque
 
# Producer-Consumer パターン（条件変数使用）
class BoundedBuffer:
    def __init__(self, capacity):
        self.buffer = deque(maxlen=capacity)
        self.capacity = capacity
        self.lock = threading.Lock()
        self.not_full = threading.Condition(self.lock)
        self.not_empty = threading.Condition(self.lock)
 
    def produce(self, item):
        with self.not_full:
            while len(self.buffer) >= self.capacity:
                self.not_full.wait()  # バッファが空くまで待機
            self.buffer.append(item)
            self.not_empty.notify()   # 消費者を起床
 
    def consume(self):
        with self.not_empty:
            while len(self.buffer) == 0:
                self.not_empty.wait()  # アイテムが来るまで待機
            item = self.buffer.popleft()
            self.not_full.notify()     # 生産者を起床
            return item
 
# 使用例
buffer = BoundedBuffer(capacity=5)
 
def producer():
    for i in range(20):
        buffer.produce(f"item-{i}")
        print(f"Produced: item-{i}")
 
def consumer(name):
    for _ in range(10):
        item = buffer.consume()
        print(f"Consumer {name} consumed: {item}")
 
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer, args=("A",))
t3 = threading.Thread(target=consumer, args=("B",))
 
t1.start(); t2.start(); t3.start()
t1.join(); t2.join(); t3.join()

// Rust: コンパイル時にデータ競合を防止
use std::thread;
use std::sync::{Arc, Mutex};
 
fn main() {
    let mut data = vec![1, 2, 3];
 
    // コンパイルエラー: 可変参照を複数スレッドに渡せない
    // thread::spawn(|| { data.push(4); });
    // thread::spawn(|| { data.push(5); });
    // エラー: `data` does not implement `Send`
    // → Rust の所有権システムがデータ競合をコンパイル時に防止
 
    // Arc<Mutex<T>> で安全に共有
    let data = Arc::new(Mutex::new(vec![1, 2, 3]));
    let d1 = Arc::clone(&data);
    let d2 = Arc::clone(&data);
 
    let h1 = thread::spawn(move || {
        d1.lock().unwrap().push(4);
    });
    let h2 = thread::spawn(move || {
        d2.lock().unwrap().push(5);
    });
 
    h1.join().unwrap();
    h2.join().unwrap();
    println!("{:?}", data.lock().unwrap()); // [1, 2, 3, 4, 5] or [1, 2, 3, 5, 4]
 
    // Send と Sync トレイト
    // Send: 値を別のスレッドに移動できる
    // Sync: 参照を別のスレッドと共有できる（&T が Send）
    //
    // ほとんどの型は Send + Sync
    // Rc<T> は Send でない（スレッドセーフでないため）
    // → Arc<T> を使う（アトミック参照カウント）
    //
    // Cell<T>, RefCell<T> は Sync でない
    // → Mutex<T>, RwLock<T> を使う
}

import threading
 
# デッドロックの例
lock_a = threading.Lock()
lock_b = threading.Lock()
 
# BAD: デッドロックの可能性
def thread1_bad():
    lock_a.acquire()
    # time.sleep(0.001)  # タイミングによりデッドロック
    lock_b.acquire()
    # 処理...
    lock_b.release()
    lock_a.release()
 
def thread2_bad():
    lock_b.acquire()  # 逆順でロック取得
    lock_a.acquire()
    # 処理...
    lock_a.release()
    lock_b.release()
 
# 回避策1: ロック順序の統一
def thread1_good():
    lock_a.acquire()  # 常に A → B の順
    lock_b.acquire()
    # 処理...
    lock_b.release()
    lock_a.release()
 
def thread2_good():
    lock_a.acquire()  # 常に A → B の順（統一）
    lock_b.acquire()
    # 処理...
    lock_b.release()
    lock_a.release()
 
# 回避策2: タイムアウト付きロック
def thread_with_timeout():
    acquired_a = lock_a.acquire(timeout=1.0)
    if not acquired_a:
        return  # タイムアウト → リトライ
 
    acquired_b = lock_b.acquire(timeout=1.0)
    if not acquired_b:
        lock_a.release()  # 保持しているロックも解放
        return  # タイムアウト → リトライ
 
    try:
        # 処理...
        pass
    finally:
        lock_b.release()
        lock_a.release()
 
# 回避策3: context manager で安全にロック管理
from contextlib import contextmanager
 
@contextmanager
def acquire_locks(*locks):
    """複数のロックをソート済みの順序で取得"""
    sorted_locks = sorted(locks, key=id)
    acquired = []
    try:
        for lock in sorted_locks:
            lock.acquire()
            acquired.append(lock)
        yield
    finally:
        for lock in reversed(acquired):
            lock.release()
 
# 使用例
def safe_thread():
    with acquire_locks(lock_a, lock_b):
        # 処理...（ロック順序は自動的に統一される）
        pass

// rayon: Rust のデータ並列処理ライブラリ
use rayon::prelude::*;
 
fn main() {
    let numbers: Vec<i64> = (0..10_000_000).collect();
 
    // 並列 map + sum
    let sum: i64 = numbers.par_iter()
        .map(|&n| n * n)
        .sum();
 
    // 並列 filter + collect
    let evens: Vec<i64> = numbers.par_iter()
        .filter(|&&n| n % 2 == 0)
        .copied()
        .collect();
 
    // 並列ソート
    let mut data = vec![5, 3, 8, 1, 9, 2, 7, 4, 6];
    data.par_sort();
 
    // カスタムスレッドプール
    let pool = rayon::ThreadPoolBuilder::new()
        .num_threads(4)
        .build()
        .unwrap();
 
    pool.install(|| {
        let result: i64 = numbers.par_iter().sum();
        println!("Sum: {}", result);
    });
 
    // 並列 for_each
    (0..100).into_par_iter().for_each(|i| {
        // 各イテレーションが並列に実行される
        println!("Processing: {}", i);
    });
}

import subprocess
import socket
import os
 
# パイプ: サブプロセスとの通信
result = subprocess.run(
    ["ls", "-la"],
    capture_output=True,
    text=True,
)
print(result.stdout)
 
# パイプチェーン: ls | grep ".py" | wc -l
p1 = subprocess.Popen(["ls"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["grep", ".py"], stdin=p1.stdout, stdout=subprocess.PIPE)
p3 = subprocess.Popen(["wc", "-l"], stdin=p2.stdout, stdout=subprocess.PIPE)
p1.stdout.close()
p2.stdout.close()
output = p3.communicate()[0]
print(f"Python files: {output.decode().strip()}")
 
# os.pipe(): 低レベルなパイプ
read_fd, write_fd = os.pipe()
 
pid = os.fork()
if pid == 0:
    # 子プロセス
    os.close(read_fd)
    os.write(write_fd, b"Hello from child!")
    os.close(write_fd)
    os._exit(0)
else:
    # 親プロセス
    os.close(write_fd)
    message = os.read(read_fd, 1024)
    os.close(read_fd)
    os.waitpid(pid, 0)
    print(f"Parent received: {message.decode()}")
 
# Unix ドメインソケット
SOCKET_PATH = "/tmp/example.sock"
 
# サーバー
def server():
    if os.path.exists(SOCKET_PATH):
        os.unlink(SOCKET_PATH)
 
    server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    server_socket.bind(SOCKET_PATH)
    server_socket.listen(1)
 
    conn, _ = server_socket.accept()
    data = conn.recv(1024)
    print(f"Server received: {data.decode()}")
    conn.send(b"Response from server")
    conn.close()
    server_socket.close()
 
# クライアント
def client():
    client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    client_socket.connect(SOCKET_PATH)
    client_socket.send(b"Hello from client")
    response = client_socket.recv(1024)
    print(f"Client received: {response.decode()}")
    client_socket.close()

package main
 
import (
    "fmt"
    "sync"
)
 
// Worker Pool パターン
func workerPool() {
    const numWorkers = 4
    jobs := make(chan int, 100)
    results := make(chan int, 100)
 
    // ワーカーの起動
    var wg sync.WaitGroup
    for w := 0; w < numWorkers; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for job := range jobs {
                result := job * job // 処理
                results <- result
            }
        }(w)
    }
 
    // ジョブの投入
    go func() {
        for i := 0; i < 20; i++ {
            jobs <- i
        }
        close(jobs)
    }()
 
    // 結果の収集
    go func() {
        wg.Wait()
        close(results)
    }()
 
    for result := range results {
        fmt.Println(result)
    }
}

// Fan-out: 1つの入力を複数のワーカーに分配
// Fan-in: 複数のワーカーの出力を1つにまとめる
 
func fanOutFanIn() {
    // 入力チャネル
    input := make(chan int)
    go func() {
        for i := 0; i < 100; i++ {
            input <- i
        }
        close(input)
    }()
 
    // Fan-out: 3つのワーカーに分配
    numWorkers := 3
    workers := make([]<-chan int, numWorkers)
    for i := 0; i < numWorkers; i++ {
        workers[i] = worker(input)
    }
 
    // Fan-in: 全ワーカーの出力を統合
    merged := merge(workers...)
 
    for result := range merged {
        fmt.Println(result)
    }
}
 
func worker(input <-chan int) <-chan int {
    output := make(chan int)
    go func() {
        defer close(output)
        for n := range input {
            output <- n * n // 処理
        }
    }()
    return output
}
 
func merge(channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    merged := make(chan int)
 
    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                merged <- v
            }
        }(ch)
    }
 
    go func() {
        wg.Wait()
        close(merged)
    }()
 
    return merged
}

// Pipeline: データが複数のステージを順に通過する
 
func pipeline() {
    // Stage 1: 数値の生成
    gen := func(nums ...int) <-chan int {
        out := make(chan int)
        go func() {
            defer close(out)
            for _, n := range nums {
                out <- n
            }
        }()
        return out
    }
 
    // Stage 2: 二乗
    square := func(in <-chan int) <-chan int {
        out := make(chan int)
        go func() {
            defer close(out)
            for n := range in {
                out <- n * n
            }
        }()
        return out
    }
 
    // Stage 3: フィルタ（偶数のみ）
    filterEven := func(in <-chan int) <-chan int {
        out := make(chan int)
        go func() {
            defer close(out)
            for n := range in {
                if n%2 == 0 {
                    out <- n
                }
            }
        }()
        return out
    }
 
    // パイプラインの接続
    result := filterEven(square(gen(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)))
 
    for v := range result {
        fmt.Println(v) // 4, 16, 36, 64, 100
    }
}