事件循环

概念定义

事件循环是一个单线程调度器，处理事件或回调的队列：等待 I/O 就绪，运行关联的处理函数直至完成，再回到队列。非阻塞 I/O 让一个线程服务大量连接，因为等待被委托给操作系统而不是占用一个线程。该模型是协作式的——任何阻塞的处理函数都会拖住整个循环。

为什么重要

线程昂贵，而多数服务端工作花在等待 I/O 而非计算上。通过不为每次等待专配一个线程，事件循环以远少的内存服务远多的连接。它还从构造上消除了大部分数据竞争，因为回调不并行运行。这正是 Node.js 与浏览器 JavaScript 能在单线程上达到高 I/O 并发的原因。

工程实践

绝不要阻塞循环：把 CPU 密集与同步文件访问卸载到工作线程或独立进程。理解阶段顺序——定时器、I/O 回调、微任务——因为它改变代码相对于 promise 完成的执行时机，并避免用冗长的回调链饿死循环。防范可能悄悄中断链条的未处理拒绝，并让回调体保持简短以约束延迟。

代码示例

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

public class CompletableFutureExample {

    static CompletableFuture<String> fetchUserName(int id) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(200);
            return "User" + id;
        });
    }

    static CompletableFuture<Integer> fetchUserScore(String name) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(300);
            return name.length() * 10;
        });
    }

    static void sleep(int ms) {
        try { Thread.sleep(ms); } catch (InterruptedException e) {}
    }

    public static void main(String[] args) throws Exception {
        CompletableFuture<String> pipeline = fetchUserName(42)
            .thenCompose(CompletableFutureExample::fetchUserScore)
            .thenApply(score -> "Score: " + score)
            .exceptionally(ex -> "Error: " + ex.getMessage());

        System.out.println(pipeline.get());

        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "Hello");
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "World");

        String combined = f1.thenCombine(f2, (a, b) -> a + " " + b).get();
        System.out.println("Combined: " + combined);

        List<CompletableFuture<String>> futures = List.of(
            CompletableFuture.supplyAsync(() -> "Task1"),
            CompletableFuture.supplyAsync(() -> "Task2"),
            CompletableFuture.supplyAsync(() -> "Task3")
        );

        CompletableFuture<Void> all = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        );
        all.thenRun(() -> System.out.println("All tasks done")).get();
    }
}

export class Semaphore {
  constructor(limit) {
    if (!Number.isInteger(limit) || limit <= 0) throw new RangeError('limit must be a positive integer');
    this.limit = limit;
    this.active = 0;
    this.waiters = [];
  }

  async acquire() {
    if (this.active < this.limit) {
      this.active += 1;
      return () => this.release();
    }
    await new Promise((resolve) => this.waiters.push(resolve));
    this.active += 1;
    return () => this.release();
  }

  release() {
    if (this.active === 0) throw new Error('release without acquire');
    this.active -= 1;
    const next = this.waiters.shift();
    if (next) {
      this.active += 1;
      next(() => this.release());
    }
  }

  async run(task) {
    const release = await this.acquire();
    try {
      return await task();
    } finally {
      release();
    }
  }
}