并发服务器：第 8 部分 - Go 语言

文章背景与核心概要

在“并发服务器”系列的第八部分中，我们探讨了 Go 语言如何应对并发编程的挑战。Go 通过提供 goroutine（一种由运行时原生管理的轻量级 M:N 调度绿色线程）避开了复杂的事件循环和基于回调的异步编程模型。

本文通过一个基础的 TCP 状态机服务器和一个更具现实意义的素数测试工作负载，演示了如何实现顺序服务器、如何通过每个客户端一个 goroutine 实现轻松扩展、如何使用带缓冲通道作为信号量来安全限制并发，以及如何实现工作池（Worker Pool）。文章最后总结道，Go 的运行时透明地处理了底层的 I/O 多路复用（如 epoll），为解决异步/等待（async/await）范式中的“函数颜色”问题提供了一种符合人体工程学的替代方案。

系列索引

注：本文假设读者对 Go 编程语言有基础了解。

顺序状态机服务器

和之前一样，我们将从第 1 部分中介绍的基础状态机协议的顺序服务器开始。

这是主函数：

func main() {
  port := "9090"
  if len(os.Args) >= 2 {
    port = os.Args[1]
  }
  log.Println("Serving on port", port)

  listener, err := net.Listen("tcp", ":"+port)
  if err != nil {
    log.Fatal("Error listening:", err)
  }
  defer listener.Close()

  for {
    conn, err := listener.Accept()
    if err != nil {
      log.Printf("Error accepting connection: %v", err)
      continue
    }
    log.Println("connection received from", conn.RemoteAddr())

    if err := server.ServeSerialProtocol(conn); err != nil {
      log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
    } else {
      log.Println("peer done", conn.RemoteAddr())
    }
  }
}

与前几部分一样，该服务器是“无限”的；除非被显式终止，否则它永远不会停止服务新连接。

这是实现单个客户端协议的函数；它接收一个代表与客户端连接的套接字的 net.Conn 值：

type processingState int

const (
  waitForMsg processingState = iota
  inMsg
)

// ServeSerialProtocol serves our serial protocol to a single TCP connection.
func ServeSerialProtocol(conn net.Conn) error {
  defer conn.Close()

  if _, err := conn.Write([]byte{'*'}); err != nil {
    return err
  }

  var state processingState = waitForMsg
  buf := make([]byte, 1024)
  for {
    n, err := conn.Read(buf)
    for _, b := range buf[:n] {
      switch state {
      case waitForMsg:
        if b == '^' {
          state = inMsg
        }
      case inMsg:
        if b == '$' {
          state = waitForMsg
        } else {
          var bb byte = byte(b) + 1
          if _, err := conn.Write([]byte{bb}); err != nil {
            return err
          }
        }
      }
    }

    // Check error after processing the bytes received along with it.
    if err != nil {
      if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) {
        return nil
      } else {
        return err
      }
    }
  }
}

每个客户端一个 Goroutine

Go 运行时并没有直接暴露操作系统线程，而是实现了自己的 M:N 调度，在操作系统线程之上运行轻量级的 goroutine。在 Go 中使用 goroutine 的成本非常低——无论是在语法和开发工作量方面，还是在系统资源方面。

下面是我们顺序协议服务器的一个版本，它通过为每个客户端启动一个 goroutine 来并发地服务客户端：

func main() {
  port := "9090"
  if len(os.Args) >= 2 {
    port = os.Args[1]
  }
  log.Println("Serving on port", port)

  listener, err := net.Listen("tcp", ":"+port)
  if err != nil {
    log.Fatal("Error listening:", err)
  }
  defer listener.Close()

  for {
    conn, err := listener.Accept()
    if err != nil {
      log.Printf("Error accepting connection: %v", err)
      continue
    }
    log.Println("connection received from", conn.RemoteAddr())

    go func() {
      if err := server.ServeSerialProtocol(conn); err != nil {
        log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
      } else {
        log.Println("peer done", conn.RemoteAddr())
      }
    }()
  }
}

在这种情况下，并发修改特别简单，因为服务器是无限循环的；没有必要等待这些 goroutine 完成（因此不需要 sync.WaitGroup）。server.ServeSerialProtocol 的参数是从封闭作用域中词法捕获的，其返回值由周围的闭包处理。

由于 goroutine 非常廉价，该服务器不太可能因为启动过多的 goroutine 而耗尽资源；事实上，它很可能会先耗尽其他资源——比如套接字的文件描述符。然而，有时限制并发程度仍然是有用的，即使是在 Go 中，我们将在接下来的章节中讨论一些实现方法。

使用信号量限制并发

在以下场景中，即使 goroutine 启动和运行成本很低，限制 Go 程序中的并发程度也是有意义的：

CPU 限制： 任务可能是计算密集型的，任何服务器的 CPU 容量本质上都是有限的。如果太多的并发 goroutine 竞争有限的 CPU，它们都将进展缓慢。让较少的任务在合理的时间内完成可能更有意义。
下游资源： 保护可能有限的下游资源，例如并发数据库连接或其他服务。例如，如果服务器必须为每个任务向其他服务发送请求，且这些服务有速率限制，则必须仔细管理并发。
安全性： 防止恶意客户端通过过度请求导致服务崩溃，从而使合法客户端无法使用服务。

让我们切换到第 4 部分中的素数测试服务器，因为这代表了更现实的工作负载。提醒一下：服务器接收数字，通过睡眠模拟阻塞，并返回 "prime" 或 "composite"。无限制的“每个客户端一个 goroutine”版本看起来与之前的代码示例几乎相同，只是 goroutine 调用了另一个函数：

go func() {
  if err := server.ServePrimeProtocol(conn); err != nil {
    log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
  } else {
    log.Println("peer done", conn.RemoteAddr())
  }
}()

其中 ServePrimeProtocol 定义如下 [1]：

// ServePrimeProtocol serves our prime protocol to a single TCP connection.
func ServePrimeProtocol(conn net.Conn) error {
  defer conn.Close()
  buf := make([]byte, 1024)

  for {
    n, readerr := conn.Read(buf)
    if n > 0 {
      // Parse the read buffer to an i64
      num, err := strconv.ParseInt(strings.TrimSpace(string(buf[:n])), 10, 64)
      if err != nil {
        return err
      }

      response := "composite"
      if isPrime(num, true) {
        response = "prime"
      }
      if _, err := conn.Write([]byte(response + "\n")); err != nil {
        return err
      }
    }

    if readerr != nil {
      if errors.Is(readerr, io.EOF) || errors.Is(readerr, net.ErrClosed) {
        return nil
      }
      return readerr
    }
  }
}

// isPrime returns true if n is prime, false otherwise. If delay is true, it
// will sleep for n milliseconds before calculating.
func isPrime(n int64, delay bool) bool {
  if delay {
    time.Sleep(time.Duration(n) * time.Millisecond)
  }

  if n < 2 {
    return false
  }
  if n%2 == 0 {
    return n == 2
  }
  for i := int64(3); i*i <= n; i += 2 {
    if n%i == 0 {
      return false
    }
  }
  return true
}

在 Go 中限制并发最简单的方法是使用计数信号量，通过通道实现：

func main() {
  port := "8070"
  if len(os.Args) >= 2 {
    port = os.Args[1]
  }
  log.Println("Serving on port", port)

  listener, err := net.Listen("tcp", ":"+port)
  if err != nil {
    log.Fatal("Error listening:", err)
  }
  defer listener.Close()

  maxConcurrency := runtime.NumCPU()
  sem := make(chan struct{}, maxConcurrency)

  for {
    conn, err := listener.Accept()
    if err != nil {
      log.Printf("Error accepting connection: %v", err)
      continue
    }
    log.Println("connection received from", conn.RemoteAddr())

    // Acquire a token from the semaphore to limit concurrency.
    sem <- struct{}{}

    go func() {
      // Return the token when done serving the connection.
      defer func() { <-sem }()

      if err := server.ServePrimeProtocol(conn); err != nil {
        log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
      } else {
        log.Println("peer done", conn.RemoteAddr())
      }
    }()
  }
}

通道 sem 用作信号量；请注意，它是一个具有最大容量的缓冲通道。通过向通道发送数据来获取令牌，通过从通道接收数据来释放令牌。当通道已满时，发送操作 sem <- struct{}{} 会阻塞，直到其他 goroutine 移除一个令牌 [2]。

通道的类型是 struct{}，这意味着“空”或“无数据”。这在 Go 中是惯用法，用于仅为了同步语义而非数据传输而使用的通道。

工作池

由于启动 goroutine 很廉价，且限制并发很容易（如上所示），对于像我们这样的服务器，通常不需要“工作池”模式。不过，它偶尔也有用处（例如当工作者需要在任务之间维护某些非平凡的状态时），因此在这里讨论一下是有价值的。

这是我们使用工作池的素数测试服务器的一个变体：

func worker(jobs <-chan net.Conn) {
  for conn := range jobs {
    if err := server.ServePrimeProtocol(conn); err != nil {
      log.Printf("Error serving %v: %v", conn.RemoteAddr(), err)
    } else {
      log.Println("peer done", conn.RemoteAddr())
    }
  }
}

func main() {
  port := "8070"
  if len(os.Args) >= 2 {
    port = os.Args[1]
  }
  log.Println("Serving on port", port)

  listener, err := net.Listen("tcp", ":"+port)
  if err != nil {
    log.Fatal("Error listening:", err)
  }
  defer listener.Close()

  // The channel is unbuffered, so it will block when there are no available
  // workers to accept a new connection.
  jobs := make(chan net.Conn)
  maxConcurrency := runtime.NumCPU()
  for i := 0; i < maxConcurrency; i++ {
    go worker(jobs)
  }

  for {
    conn, err := listener.Accept()
    if err != nil {
      log.Printf("Error accepting connection: %v", err)
      continue
    }
    log.Println("connection received from", conn.RemoteAddr())

    jobs <- conn
  }
}

启动固定数量的工作者 goroutine；这些 goroutine 都从同一个通道接收“作业”。在 Accept 循环中，每个客户端连接作为新作业发送到此通道，并由下一个可用的工作者获取。如前所述，通常会在某处看到 sync.WaitGroup 以确保 goroutine 的干净关闭，但在我们的例子中，由于服务器从不退出，因此没有必要。

异步？

程序员在 Go 中是否必须求助于异步/事件驱动编程？根据我的经验，几乎不需要。Go 从底层设计上就适合大规模并发；goroutine 创建非常廉价，内存占用极小，切换速度非常快——所有这些都在用户空间完成。我在 2018 年进行的测量显示，切换时间约为 170 纳秒，而 Linux 线程的切换时间为 1-2 微秒。

此外，Go 在底层已经使用了像 epoll 这样的事件驱动循环进行 I/O。等待 I/O（如套接字）的 goroutine 实际上被“挂起”，除了它们的小内存占用外不消耗任何资源；当它们的 I/O 描述符就绪时，它们会被 Go 的运行时唤醒——这与异步编程的工作方式非常相似！

话虽如此，当同时处理 数百万 个流时，有些人确实会尝试通过直接的异步编程来进一步挖掘资源。我只想说，这种情况非常罕见，绝大多数用户永远不需要这样做。

结论

2018 年，我写了一篇题为 Go hits the concurrency nail right on the head 的文章，在经过几年的积极编码后，我完全支持这一说法。Go 对于并发程序来说非常强大且符合人体工程学；虽然其他环境在库中费尽心思实现 async-await 风格的事件循环，但在 Go 中，它已经内置在核心语言和运行时中。你想要带有非常轻量级绿色线程的事件驱动 I/O，并且可以执行阻塞任务而无需担心函数颜色问题吗？Go 已经为你准备好了。

代码

本文的所有代码均可在 GitHub 上找到。

脚注

细心的读者会注意到这段代码的两个问题：
该协议假设完整的数字是在一次 conn.Read 调用中从套接字读取的，并且不同数字之间没有帧或分隔。
素数检查循环使用 i*i，对于大数字可能会溢出。

这些问题在之前部分中 C、Python、JavaScript 和 Rust 版本的素数服务器中是一致的，因为我的重点是展示并发点的最简单代码。

练习： 请注意，我们的信号量保护了整个 ServePrimeProtocol，这意味着恶意（或无能的）客户端可以连接并空闲而不发送任何请求，从而将我们的并发容量减少 1。调整代码，将信号量移动到素数计算本身周围，以便仅限制此部分。