异步 I/O

概念概述

异步 I/O（Asynchronous I/O）指的是：异步 I/O 发起操作后立即继续而不阻塞，稍后通知完成，使一个线程可服务多个套接字或文件操作，是高性能服务的基础。

从更严格的定义看，Asynchronous input and output is a model in which a program initiates an operation and receives a completion indication later, instead of blocking the calling thread until the operation finishes. 该概念在工程实践中的价值，在于它能帮助团队用统一的语言描述一个具体的质量属性或设计约束，从而在评审与协作中减少歧义。

为什么需要它

It is the basis of high-throughput servers, event loops, and runtimes that must juggle thousands of concurrent network connections. 它与 Goroutine and Task Leaks、I/O Multiplexing with epoll and kqueue、Atomic Operations 等概念经常一起出现，用于解决相近的工程问题。理解它的适用场景，有助于在多种备选方案之间做出有依据的取舍，而不是凭直觉选择技术方案。

常见误解

• asynchronous IO is the same as non-blocking IO
• asynchronous operations necessarily run on other threads

需要避免的反模式

• calling synchronous file APIs inside an async handler
• assuming all filesystems support true asynchronous IO equally

实践要点

1. Non-blocking IO returns immediately when an operation would block and the program polls readiness, whereas asynchronous IO hands the whole operation to the kernel or runtime and reports completion.
2. Completion can arrive via callbacks, futures, or an event loop, and the thread that initiated the operation is free in the meantime.
3. Mixing synchronous calls into an asynchronous pipeline defeats the model, because one blocking call stalls every task scheduled on that event loop.

参考来源

https://en.wikipedia.org/wiki/Asynchronous_I/O

代码示例

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();
    }
}