Hello Golang

第4章高级网络编程

4.1 非阻塞 I/O

非阻塞 I/O 的目标是读写不把线程卡死，常见的做法有三种：select、poll、epoll。

4.1.1 基于 select 的 I/O 多路复用

select 是早期的 I/O 多路复用技术，同时监控多个文件描述符，看哪些可以读写。它的限制是文件描述符数量有限、效率较低。

示例代码：

package main

import (
	"fmt"
	"net"
	"os"
	"syscall"
)

func main() {
	ln, err := net.Listen("tcp", ":8080")
	if err != nil {
		fmt.Println("Error listening:", err)
		os.Exit(1)
	}
	defer ln.Close()

	// 获取监听 socket 的文件描述符（File() 会返回 dup 出来的 *os.File 和错误）
	tcpLn := ln.(*net.TCPListener)
	file, err := tcpLn.File()
	if err != nil {
		fmt.Println("Error getting file:", err)
		os.Exit(1)
	}
	defer file.Close()
	fd := int(file.Fd())

	for {
		readfds := &syscall.FdSet{}
		// 注意：Go 在 Linux 上没有提供 FdSet.Set/IsSet 等便捷方法，
		// 需要手动设置位图：Bits 是 [16]int64，第 fd 位为 1
		readfds.Bits[fd/64] |= 1 << (uint(fd) % 64)

		// select 的最后一个参数是超时时间（Timeval），nil 表示永久阻塞
		timeout := &syscall.Timeval{Sec: 1, Usec: 0}
		n, err := syscall.Select(fd+1, readfds, nil, nil, timeout)
		if err != nil {
			fmt.Println("Error in select:", err)
			continue
		}
		if n == 0 {
			continue // 超时，没有就绪的描述符
		}

		// 同样手动检查位图
		if readfds.Bits[fd/64]&(1<<(uint(fd)%64)) != 0 {
			conn, err := ln.Accept()
			if err != nil {
				fmt.Println("Error accepting:", err)
				continue
			}

			go handleConnection(conn)
		}
	}
}

func handleConnection(conn net.Conn) {
	defer conn.Close()
	buf := make([]byte, 1024)
	for {
		n, err := conn.Read(buf)
		if err != nil {
			fmt.Println("Error reading:", err)
			return
		}
		fmt.Println("Received data:", string(buf[:n]))
	}
}

4.1.2 基于 poll 的 I/O 多路复用

poll 是对 select 的改进，它消除了文件描述符数量的限制，并且效率较高。poll 使用一个结构数组来监控文件描述符的事件。

示例代码：

注意：Go 的 syscall 包在 Linux/amd64 上没有导出 PollFd、POLLIN 和 Poll，使用 poll 需要 golang.org/x/sys/unix 包（go get golang.org/x/sys）：

package main

import (
	"fmt"
	"net"
	"os"

	"golang.org/x/sys/unix"
)

func main() {
	ln, err := net.Listen("tcp", ":8080")
	if err != nil {
		fmt.Println("Error listening:", err)
		os.Exit(1)
	}
	defer ln.Close()

	tcpLn := ln.(*net.TCPListener)
	file, err := tcpLn.File()
	if err != nil {
		fmt.Println("Error getting file:", err)
		os.Exit(1)
	}
	defer file.Close()
	fd := int(file.Fd())

	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}

	for {
		n, err := unix.Poll(fds, -1) // -1 表示永久阻塞
		if err != nil {
			fmt.Println("Error in poll:", err)
			continue
		}
		if n == 0 {
			continue
		}

		if fds[0].Revents&unix.POLLIN != 0 {
			conn, err := ln.Accept()
			if err != nil {
				fmt.Println("Error accepting:", err)
				continue
			}

			go handleConnection(conn)
		}
	}
}

func handleConnection(conn net.Conn) {
	defer conn.Close()
	buf := make([]byte, 1024)
	for {
		n, err := conn.Read(buf)
		if err != nil {
			fmt.Println("Error reading:", err)
			return
		}
		fmt.Println("Received data:", string(buf[:n]))
	}
}

4.1.3 基于 epoll 的 I/O 多路复用

epoll 是 Linux 平台上更高效的 I/O 多路复用机制，适用于大规模并发连接。epoll 提供了边缘触发（ET）和水平触发（LT）两种模式，通常与事件驱动编程结合使用。

示例代码：

package main

import (
	"fmt"
	"net"
	"os"
	"syscall"
)

func main() {
	ln, err := net.Listen("tcp", ":8080")
	if err != nil {
		fmt.Println("Error listening:", err)
		os.Exit(1)
	}
	defer ln.Close()

	tcpLn := ln.(*net.TCPListener)
	listenFile, err := tcpLn.File()
	if err != nil {
		fmt.Println("Error getting file:", err)
		os.Exit(1)
	}
	defer listenFile.Close()
	listenFd := int(listenFile.Fd())
	epollFd, err := syscall.EpollCreate1(0)
	if err != nil {
		fmt.Println("Error creating epoll:", err)
		os.Exit(1)
	}
	defer syscall.Close(epollFd)

	event := syscall.EpollEvent{Events: syscall.EPOLLIN, Fd: int32(listenFd)}
	err = syscall.EpollCtl(epollFd, syscall.EPOLL_CTL_ADD, listenFd, &event)
	if err != nil {
		fmt.Println("Error adding listen fd to epoll:", err)
		os.Exit(1)
	}

	events := make([]syscall.EpollEvent, 10)
	for {
		n, err := syscall.EpollWait(epollFd, events, -1)
		if err != nil {
			fmt.Println("Error in epoll wait:", err)
			continue
		}

		for i := 0; i < n; i++ {
			if int(events[i].Fd) == listenFd {
				conn, err := ln.Accept()
				if err != nil {
					fmt.Println("Error accepting:", err)
					continue
				}

				connFile, err := conn.(*net.TCPConn).File()
				if err != nil {
					fmt.Println("Error getting conn file:", err)
					conn.Close()
					continue
				}
				connFd := int(connFile.Fd())
				event := syscall.EpollEvent{Events: syscall.EPOLLIN, Fd: int32(connFd)}
				err = syscall.EpollCtl(epollFd, syscall.EPOLL_CTL_ADD, connFd, &event)
				if err != nil {
					fmt.Println("Error adding conn fd to epoll:", err)
					conn.Close()
					continue
				}
			} else {
				go handleConnection(int(events[i].Fd))
			}
		}
	}
}

func handleConnection(fd int) {
	conn := os.NewFile(uintptr(fd), "")
	defer conn.Close()

	buf := make([]byte, 1024)
	for {
		n, err := conn.Read(buf)
		if err != nil {
			fmt.Println("Error reading:", err)
			return
		}
		fmt.Println("Received data:", string(buf[:n]))
	}
}

4.2 事件驱动编程

事件驱动编程是一种编程范式，通过事件循环和回调机制处理异步事件。事件驱动编程特别适合处理 I/O 密集型和高并发任务。

4.2.1 事件循环

事件循环是事件驱动编程的核心，它不断地等待和分发事件。事件循环通常与非阻塞 I/O 和 I/O 多路复用技术结合使用，以实现高效的事件处理。

示例代码：

package main

import (
	"fmt"
	"net"
	"time"
)

type EventLoop struct {
	events chan func()
}

func NewEventLoop() *EventLoop {
	return &EventLoop{events: make(chan func(), 1024)}
}

func (el *EventLoop) AddEvent(event func()) {
	el.events <- event
}

func (el *EventLoop) Run() {
	for event := range el.events {
		event()
	}
}

func main() {
	loop := NewEventLoop()
	go loop.Run()

	ln, err := net.Listen("tcp", ":8080")
	if err != nil {
		fmt.Println("Error listening:", err)
		return
	}
	defer ln.Close()

	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				fmt.Println("Error accepting:", err)
				continue
			}
			handleConnection(loop, conn)
		}
	}()

	time.Sleep(10 * time.Second)
}

func handleConnection(loop *EventLoop, conn net.Conn) {
	buf := make([]byte, 1024)
	go func() {
		for {
			n, err := conn.Read(buf)
			if err != nil {
				fmt.Println("Error reading:", err)
				conn.Close()
				return
			}
			loop.AddEvent(func() {
				fmt.Println("Received data:", string(buf[:n]))
			})
		}
	}()
}

4.2.2 回调机制

回调机制是事件驱动编程的重要组成部分，通过在事件发生时调用预定义的回调函数，实现事件的异步处理。

示例代码：

package main

import (
	"fmt"
	"net"
	"time"
)

type EventHandler func(data []byte)

func main() {
	handler := func(data []byte) {
		fmt.Println("Received data:", string(data))
	}

	ln, err := net.Listen("tcp", ":8080")
	if err != nil {
		fmt.Println("Error listening:", err)
		return
	}
	defer ln.Close()

	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				fmt.Println("Error accepting:", err)
				continue
			}
			handleConnection(conn, handler)
		}
	}()

	time.Sleep(10 * time.Second)
}

func handleConnection(conn net.Conn, handler EventHandler) {
	buf := make([]byte, 1024)
	go func() {
		for {
			n, err := conn.Read(buf)
			if err != nil {
				fmt.Println("Error reading:", err)
				conn.Close()
				return
			}
			handler(buf[:n])
		}


	}()
}

4.3 异步 I/O

异步 I/O 是一种在 I/O 操作完成之前立即返回的编程技术，通过回调函数或事件通知机制处理 I/O 结果。异步 I/O 可以提高应用程序的并发性和响应速度。

4.3.1 基于 IOCP 的异步 I/O

IOCP（I/O Completion Ports）是 Windows 平台上的一种高效异步 I/O 机制，通过 I/O 完成端口和线程池实现高并发 I/O 操作。

示例代码（伪代码）：

// 这是一个 IOCP 示例的伪代码，实际实现需要使用 Windows API
#include <windows.h>
#include <stdio.h>

void CALLBACK IoCompletionCallback(DWORD errorCode, DWORD numberOfBytesTransferred, LPOVERLAPPED overlapped) {
    printf("I/O operation completed with %d bytes transferred.\n", numberOfBytesTransferred);
}

int main() {
    HANDLE iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);

    HANDLE file = CreateFile("example.txt", GENERIC_READ, 0, NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);
    CreateIoCompletionPort(file, iocp, (ULONG_PTR)file, 0);

    char buffer[1024];
    OVERLAPPED overlapped = {0};
    ReadFile(file, buffer, sizeof(buffer), NULL, &overlapped);

    DWORD numberOfBytesTransferred;
    ULONG_PTR completionKey;
    LPOVERLAPPED completedOverlapped;
    GetQueuedCompletionStatus(iocp, &numberOfBytesTransferred, &completionKey, &completedOverlapped, INFINITE);

    IoCompletionCallback(0, numberOfBytesTransferred, completedOverlapped);

    CloseHandle(file);
    CloseHandle(iocp);
    return 0;
}

4.3.2 基于 io_uring 的异步 I/O

io_uring 是 Linux 内核中的一种高性能异步 I/O 机制，通过环形缓冲区实现高效的 I/O 操作提交和完成通知。

示例代码（伪代码）：

// 这是一个 IO_uring 示例的伪代码，实际实现需要使用 Linux IO_uring API
#include <liburing.h>
#include <stdio.h>
#include <fcntl.h>
#include <unistd.h>

int main() {
    struct io_uring ring;
    io_uring_queue_init(32, &ring, 0);

    int fd = open("example.txt", O_RDONLY | O_DIRECT);
    char buffer[4096];

    struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
    io_uring_prep_read(sqe, fd, buffer, sizeof(buffer), 0);
    io_uring_submit(&ring);

    struct io_uring_cqe *cqe;
    io_uring_wait_cqe(&ring, &cqe);

    printf("I/O operation completed with %d bytes read.\n", cqe->res);
    io_uring_cqe_seen(&ring, cqe);

    close(fd);
    io_uring_queue_exit(&ring);
    return 0;
}

4.4 网络协议解析

这一节看三种协议的解析方式：HTTP、WebSocket，以及按自己格式走的自定义协议。

4.4.1 HTTP 协议解析

HTTP（Hypertext Transfer Protocol）是一种用于分布式、协作、超媒体信息系统的应用层协议。HTTP 是 Web 的基础，通过请求和响应机制进行通信。

示例代码：

package main

import (
	"fmt"
	"net/http"
)

func handler(w http.ResponseWriter, r *http.Request) {
	fmt.Fprintf(w, "Hello, World!")
}

func main() {
	http.HandleFunc("/", handler)
	http.ListenAndServe(":8080", nil)
}

4.4.2 Websocket 协议解析

WebSocket 是一种全双工通信协议，通过单个 TCP 连接实现客户端和服务器之间的实时数据交换。

示例代码：

package main

import (
	"fmt"
	"net/http"

	"github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
}

func handler(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		fmt.Println("Error upgrading to websocket:", err)
		return
	}
	defer conn.Close()

	for {
		messageType, message, err := conn.ReadMessage()
		if err != nil {
			fmt.Println("Error reading message:", err)
			return
		}
		fmt.Printf("Received: %s\n", message)

		if err := conn.WriteMessage(messageType, message); err != nil {
			fmt.Println("Error writing message:", err)
			return
		}
	}
}

func main() {
	http.HandleFunc("/", handler)
	http.ListenAndServe(":8080", nil)
}

4.4.3 自定义协议

自定义协议按自己定的格式走，下面这个例子最简单：一行一条消息，收到什么就回一个 ACK。

示例代码：

package main

import (
	"bufio"
	"fmt"
	"net"
	"strings"
)

func handleConnection(conn net.Conn) {
	defer conn.Close()
	reader := bufio.NewReader(conn)
	for {
		message, err := reader.ReadString('\n')
		if err != nil {
			fmt.Println("Error reading:", err)
			return
		}
		message = strings.TrimSpace(message)
		fmt.Printf("Received: %s\n", message)

		response := "ACK\n"
		_, err = conn.Write([]byte(response))
		if err != nil {
			fmt.Println("Error writing:", err)
			return
		}
	}
}

func main() {
	ln, err := net.Listen("tcp", ":8080")
	if err != nil {
		fmt.Println("Error listening:", err)
		return
	}
	defer ln.Close()

	for {
		conn, err := ln.Accept()
		if err != nil {
			fmt.Println("Error accepting:", err)
			continue
		}
		go handleConnection(conn)
	}
}

结论

select、poll、epoll 解决的是"一个线程盯多个连接"，事件循环和回调解决的是"事件到了由谁来接"，IOCP 和 io_uring 属于异步提交加完成通知的路子，协议解析则决定字节流里怎么切出一条条消息。Go 的 net 和 net/http 已经把这些封装在标准库里，这些底层写法是为了看懂它们为什么这么设计。

集成开发环境

条件语句

循环语句

跳转语句

创建型模式

结构型模式

行为模式

数组

链表

树

排序

消息队列系统

第4章高级网络编程

4.1 非阻塞 I/O

4.1.1 基于 select 的 I/O 多路复用

4.1.2 基于 poll 的 I/O 多路复用

4.1.3 基于 epoll 的 I/O 多路复用

4.2 事件驱动编程

4.2.1 事件循环

4.2.2 回调机制

4.3 异步 I/O

4.3.1 基于 IOCP 的异步 I/O

4.3.2 基于 io_uring 的异步 I/O

4.4 网络协议解析

4.4.1 HTTP 协议解析

4.4.2 Websocket 协议解析

4.4.3 自定义协议

结论

参考资料

同层页面

相关页面

第4章 高级网络编程 ​

4.1 非阻塞 I/O ​

4.1.1 基于 select 的 I/O 多路复用 ​

4.1.2 基于 poll 的 I/O 多路复用 ​

4.1.3 基于 epoll 的 I/O 多路复用 ​

4.2 事件驱动编程 ​

4.2.1 事件循环 ​

4.2.2 回调机制 ​

4.3 异步 I/O ​

4.3.1 基于 IOCP 的异步 I/O ​

4.3.2 基于 io_uring 的异步 I/O ​

4.4 网络协议解析 ​

4.4.1 HTTP 协议解析 ​

4.4.2 Websocket 协议解析 ​

4.4.3 自定义协议 ​

结论 ​

参考资料 ​

同层页面 ​

相关页面 ​

第4章高级网络编程

4.1 非阻塞 I/O

4.1.1 基于 select 的 I/O 多路复用

4.1.2 基于 poll 的 I/O 多路复用

4.1.3 基于 epoll 的 I/O 多路复用

4.2 事件驱动编程

4.2.1 事件循环

4.2.2 回调机制

4.3 异步 I/O

4.3.1 基于 IOCP 的异步 I/O

4.3.2 基于 io_uring 的异步 I/O

4.4 网络协议解析

4.4.1 HTTP 协议解析

4.4.2 Websocket 协议解析

4.4.3 自定义协议

结论

参考资料

同层页面

相关页面