如何处理并发, Green Thread? Async? OS Thread?

// 同步阻塞示例
fn serve_sync() {
    pour();           // 15秒
    sleep(100);       // ← CPU在这里干等着,什么都不做
    top_up();         // 5秒
    serve();
}

// 异步非阻塞示例
async fn serve_async() {
    pour().await;           // 15秒,实际工作
    tokio::spawn(async {    // 启动一个后台任务等待沉淀
        sleep(100).await;   // 在等待期间,当前线程可以处理其他订单
        top_up().await;     // 5秒,实际工作
        serve().await;
    });
    // 立即返回,可以接下一个订单
}

pthread_t thread;
pthread_create(&thread, NULL, task_function, arg);
pthread_join(thread, NULL);

Thread thread = new Thread(() -> {
    // 任务代码
});
thread.start();
thread.join();

import threading
thread = threading.Thread(target=task_function, args=(arg,))
thread.start()
thread.join()

let handle = std::thread::spawn(|| {
    // 任务代码
});
handle.join().unwrap();

// 用户写的代码
conn.Read(buf)  // 看起来是阻塞调用

// Cgo 调用
C.some_c_function()

// 创建100万个并发任务
for i := 0; i < 1000000; i++ {
    go func(id int) {
        result := doWork(id)  // 看起来是阻塞，实际非阻塞
        fmt.Println(result)
    }(i)
}
// 每个 Goroutine 只占用约2KB内存

# 创建一个轻量级进程
spawn(fn ->
  result = do_work()
  IO.puts(result)
end)
# Erlang VM 可以运行数百万个进程

// JDK 21+ 的虚拟线程
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 1000000; i++) {
        executor.submit(() -> {
            // 虚拟线程，调度在少量平台线程上
            String result = blockingIO();
            System.out.println(result);
        });
    }
}

fiber = Fiber.new do
  puts "Fiber started"
  Fiber.yield  # 显式让出控制权
  puts "Fiber resumed"
end
fiber.resume  # 启动/恢复 Fiber

// 主线程执行
console.log("1. 开始");
 
// 发起异步操作，注册回调
readFile("data.txt", function callback(data) {
    console.log("3. 文件内容:", data);
});
 
console.log("2. 继续执行");
 
// 输出顺序: 1 → 2 → 3

fetchUser(userId, function(user) {
    fetchPosts(user.id, function(posts) {
        fetchComments(posts[0].id, function(comments) {
            fetchReplies(comments[0].id, function(replies) {
                // 已经嵌套4层了...看着头疼吧
                console.log(replies);
            });
        });
    });
});

// 经典的 Node.js 回调风格
const fs = require('fs');
 
fs.readFile('file.txt', 'utf8', (err, data) => {
    if (err) {
        console.error('Error:', err);
        return;
    }
    console.log('Content:', data);
});
 
// HTTP 请求回调
const http = require('http');
http.get('http://api.example.com', (res) => {
    let data = '';
    res.on('data', (chunk) => { data += chunk; });
    res.on('end', () => {
        console.log('Response:', data);
    });
});

from twisted.internet import reactor, defer
 
def on_success(result):
    print(f"Success: {result}")
 
def on_error(failure):
    print(f"Error: {failure}")
 
# 注册回调
d = defer.Deferred()
d.addCallback(on_success)
d.addErrback(on_error)
 
# 某处触发回调
d.callback("Operation completed")

CompletableFuture.supplyAsync(() -> {
    return fetchData();  // 异步执行
}).thenAccept(data -> {
    System.out.println("Data: " + data);  // 回调处理
}).exceptionally(ex -> {
    System.err.println("Error: " + ex);
    return null;
});

// libuv 的异步文件读取
uv_fs_t req;
uv_fs_read(loop, &req, file, &buf, 1, -1, on_read_callback);
 
void on_read_callback(uv_fs_t* req) {
    if (req->result < 0) {
        fprintf(stderr, "Read error\n");
    } else {
        printf("Read %ld bytes\n", req->result);
    }
    uv_fs_req_cleanup(req);
}

async fn fetch_and_process() -> String {
    let response = http_get("http://api.com").await;  // 等待点1
    let data = parse(response).await;                 // 等待点2
    format!("Result: {}", data)
}
```：
编译器会把它转换成一个状态机:
Future/Promise 代表一个"尚未完成的计算"的对象，而 async/await 则是让你用同步的代码风格编写异步逻辑的语法糖.与 Green Thread 最关键的区别在于：Async/Await **没有独立的栈**，所有状态都保存在堆上的状态机对象中，这使得每个任务的内存占用可以降到几百字节
```rust
enum FetchAndProcessState {
    Start,
    WaitingForHttp { /* 保存的局部变量 */ },
    WaitingForParse { response: String },
    Done,
}
    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<String> {
        loop {
            match self.state {
                Start => {
                    // 发起 HTTP 请求
                    let fut = http_get("http://api.com");
                    match fut.poll(cx) {
                        Poll::Pending => {
                            self.state = WaitingForHttp;
                            return Poll::Pending;  // 告诉运行时：我还没完成
                        }
                        Poll::Ready(response) => {
                            self.state = WaitingForParse { response };
                        }
                    }
                }
                WaitingForParse { response } => {
                    let fut = parse(response);
                    match fut.poll(cx) {
                        Poll::Pending => return Poll::Pending,
                        Poll::Ready(data) => {
                            return Poll::Ready(format!("Result: {}", data));
                        }
                    }
                }
                _ => panic!("Invalid state"),
            }
        }
    }
}

async fn long_compute() {
    // 要是这里有个超耗时的计算循环
    for i in 0..1000000000 {
        // 没有 .await 点！
        compute(i);
    }
    // 运行时没法中途打断
}

use tokio;
 
#[tokio::main]
async fn main() {
    let task1 = tokio::spawn(async {
        let data = fetch_data().await;
        process(data).await
    });
    
    let task2 = tokio::spawn(async {
        let result = compute().await;
        save(result).await。
 
**对比 Go**：编译器在每个函数调用点塞了调度检查，可以随时抢占。
    // 并发等待多个任务
    let (r1, r2) = tokio::join!(task1, task2);
}
 
async fn fetch_data() -> String {
    // 编译成状态机
    tokio::time::sleep(Duration::from_secs(1)).await;
    "data".to_string()
}

// async 函数返回 Promise
async function fetchUserData(userId) {
    const user = await fetch(`/api/users/${userId}`);
    const posts = await fetch(`/api/posts/${user.id}`);
    return { user, posts };
}
 
// Promise 链式调用(底层实现)
function fetchUserDataPromise(userId) {
    return fetch(`/api/users/${userId}`)
        .then(user => fetch(`/api/posts/${user.id}`)
        .then(posts => ({ user, posts })));
}
 
// 并发执行多个异步任务
const [users, posts, comments] = await Promise.all([
    fetch('/api/users'),
    fetch('/api/posts'),
    fetch('/api/comments')
]);

import asyncio
 
async def fetch_data(url):
    # await 让出控制权
    response = await http_client.get(url)
    data = await response.json()
    return data
 
async def main():
    # 并发运行多个任务
    results = await asyncio.gather(
        fetch_data("http://api1.com"),
        fetch_data("http://api2.com"),
        fetch_data("http://api3.com")
    )
    print(results)
 
# 运行事件循环
asyncio.run(main())

public async Task<string> FetchAndProcessAsync() {
    // async 方法返回 Task<T>
    var response = await httpClient.GetAsync("http://api.com");
    var content = await response.Content.ReadAsStringAsync();
    var result = await ProcessAsync(content);
    return result;
}
 
// 并发执行
var task1 = FetchDataAsync("url1");
var task2 = FetchDataAsync("url2");
await Task.WhenAll(task1, task2);

suspend fun fetchData(): String {
    delay(1000)  // 挂起点
    return "data"
}
 
fun main() = runBlocking {
    // 启动多个协程
    val job1 = launch {
        val data = fetchData()
        println(data)
    }
    
    val job2 = launch {
        val result = compute()
        println(result)
    }
    
    job1.join()
    job2.join()
}