1
package main
2

3
import (
4
    "log"
5
    "net"
6
)
7

8
func handleConn(conn net.Conn) {
9
    defer conn.Close()
10
    buf := make([]byte, 4096)
11

12
    for {
13
        // 看起来是阻塞读，实际 goroutine 会在 I/O 等待时挂起
14
        // 底层由 runtime 的 epoll 事件循环驱动
15
        n, err := conn.Read(buf)
16
        if err != nil {
17
            log.Printf("连接 %s 读取结束: %v", conn.RemoteAddr(), err)
18
            return
19
        }
20
        // 回显：把收到的数据原样写回
21
        if _, err := conn.Write(buf[:n]); err != nil {
22
            log.Printf("连接 %s 写入失败: %v", conn.RemoteAddr(), err)
23
            return
24
        }
25
    }
26
}
27

28
func main() {
29
    ln, err := net.Listen("tcp", ":8080")
30
    if err != nil {
31
        log.Fatal(err)
32
    }
33
    defer ln.Close()
34
    log.Println("服务启动在 :8080")
35

36
    for {
37
        conn, err := ln.Accept()
38
        if err != nil {
39
            log.Printf("接受连接失败: %v", err)
40
            continue
41
        }
42
        // 每个连接一个 goroutine，看似一连接一线程
43
        // 实际由 Go 调度器在少量系统线程上复用
44
        go handleConn(conn)
45
    }
46
}

1
package main
2

3
import (
4
    "log"
5
    "golang.org/x/sys/unix"
6
)
7

8
func main() {
9
    // 创建 epoll 实例
10
    epfd, err := unix.EpollCreate1(0)
11
    if err != nil {
12
        log.Fatal(err)
13
    }
14
    defer unix.Close(epfd)
15

16
    // 创建监听 socket
17
    lfd, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM|unix.SOCK_NONBLOCK, 0)
18
    if err != nil {
19
        log.Fatal(err)
20
    }
21
    defer unix.Close(lfd)
22

23
    // 注册监听 fd 的可读事件（新连接到来）
24
    event := &unix.EpollEvent{
25
        Events: unix.EPOLLIN,
26
        Fd:     int32(lfd),
27
    }
28
    if err := unix.EpollCtl(epfd, unix.EPOLL_CTL_ADD, lfd, event); err != nil {
29
        log.Fatal(err)
30
    }
31

32
    events := make([]unix.EpollEvent, 128)
33
    for {
34
        // Reactor 核心：等待事件就绪
35
        nevents, err := unix.EpollWait(epfd, events, -1)
36
        if err != nil {
37
            continue
38
        }
39
        // 分发就绪事件给对应的处理器
40
        for i := 0; i < nevents; i++ {
41
            ev := events[i]
42
            if int(ev.Fd) == lfd {
43
                // 监听 fd 就绪 → 接受新连接
44
                nfd, _, err := unix.Accept(lfd)
45
                if err != nil {
46
                    continue
47
                }
48
                // 把新连接也加进 epoll
49
                connEvent := &unix.EpollEvent{
50
                    Events: unix.EPOLLIN,
51
                    Fd:     int32(nfd),
52
                }
53
                unix.EpollCtl(epfd, unix.EPOLL_CTL_ADD, nfd, connEvent)
54
                log.Printf("新连接: fd=%d", nfd)
55
            } else {
56
                // 普通 fd 就绪 → 读取数据
57
                buf := make([]byte, 512)
58
                n, err := unix.Read(int(ev.Fd), buf)
59
                if err != nil || n == 0 {
60
                    // 连接关闭或出错，从 epoll 移除
61
                    unix.EpollCtl(epfd, unix.EPOLL_CTL_DEL, int(ev.Fd), nil)
62
                    unix.Close(int(ev.Fd))
63
                    continue
64
                }
65
                log.Printf("fd=%d 收到 %d 字节", ev.Fd, n)
66
            }
67
        }
68
    }
69
}

1
import { EventEmitter } from "events";
2

3
// 模拟多路复用器：监听多个 fd 的就绪状态
4
class Demultiplexer extends EventEmitter {
5
  private fds: Map<number, string> = new Map();
6

7
  // 注册对某个 fd 的兴趣
8
  register(fd: number, interest: string): void {
9
    this.fds.set(fd, interest);
10
  }
11

12
  // 模拟事件就绪通知（实际由 epoll/kqueue 驱动）
13
  notifyReady(fd: number, event: string): void {
14
    this.emit("ready", { fd, event });
15
  }
16
}
17

18
// 具体事件处理器
19
class ReadHandler {
20
  onEvent(fd: number, event: string): void {
21
    if (event === "readable") {
22
      // Reactor 特征：收到就绪通知后，应用自己执行 I/O
23
      const data = this.doRead(fd);
24
      console.log(`fd=${fd} 读取到: ${data}`);
25
    }
26
  }
27

28
  private doRead(fd: number): string {
29
    // 模拟非阻塞读取
30
    return `来自 fd=${fd} 的数据`;
31
  }
32
}
33

34
// Reactor 事件循环
35
const demux = new Demultiplexer();
36
const handler = new ReadHandler();
37

38
demux.on("ready", ({ fd, event }) => {
39
  handler.onEvent(fd, event);
40
});
41

42
demux.register(3, "readable");
43
demux.notifyReady(3, "readable"); // 输出: fd=3 读取到: 来自 fd=3 的数据

1
import { open } from "fs/promises";
2

3
// Proactor 风格：发起异步操作，等完成通知
4
async function proactorStyle(): Promise<void> {
5
  // 发起异步读取 → 操作系统执行 → 完成后 Promise resolve
6
  // 应用完全不参与 I/O 执行过程
7
  const handle = await open("/tmp/test.txt", "r");
8
  const buffer = Buffer.alloc(1024);
9

10
  // 异步读取：内核完成数据拷贝，通过 Promise 通知完成
11
  const { bytesRead } = await handle.read(buffer, 0, 1024, 0);
12

13
  // 收到完成通知时，数据已经在 buffer 里了
14
  console.log(`读取完成，${bytesRead} 字节: ${buffer.toString("utf8", 0, bytesRead)}`);
15
  await handle.close();
16
}
17

18
// 完成处理器：对应 Proactor 的 Completion Handler
19
async function completionHandler(
20
  filePath: string,
21
  onComplete: (data: string) => void
22
): Promise<void> {
23
  const handle = await open(filePath, "r");
24
  const buffer = Buffer.alloc(4096);
25
  const { bytesRead } = await handle.read(buffer, 0, 4096, 0);
26
  // I/O 完成后调用回调，数据已经就绪
27
  onComplete(buffer.toString("utf8", 0, bytesRead));
28
  await handle.close();
29
}
30

31
// 使用
32
completionHandler("/tmp/test.txt", (data) => {
33
  console.log(`完成处理: ${data}`);
34
});