标准库 - Tony Bai

标签标准库下的文章

http.Client的连接行为控制详解

四月 2, 2021
0 条评论

1. http包默认客户端

Go语言以“自带电池”闻名，很多开发者对Go自带的功能丰富的标准库喜爱有加。而在Go标准库中，net/http包又是最受欢迎和最常用的包之一，我们用几行代码就能生成一个支持大并发、性能中上的http server。而http.Client也是用途最为广泛的http客户端，其性能也可以满足多数情况下的需求。知名女gopherJaana Dogan开源的类apache ab的http性能测试工具hey也是直接使用的http.Client，而没有用一些性能更好的第三方库（比如：fasthttp)。

使用http包实现http客户端的最简单方法如下(来自http包的官方文档)：

resp, err := http.Get("http://example.com/")
...
resp, err := http.Post("http://example.com/upload", "image/jpeg", &buf)
...

注：别忘了在Get或Post成功后，调用defer resp.Body.Close()。

在http包的Get和Post函数背后，真正完成http客户端操作的是http包原生内置的DefaultClient：

// $GOROOT/src/net/http/client.go
// DefaultClient is the default Client and is used by Get, Head, and Post.
var DefaultClient = &Client{}

下面是一个使用DefaultClient的例子，我们先来创建一个特殊的http server：

// github.com/bigwhite/experiments/blob/master/http-client/default-client/server.go

package main

import (
    "log"
    "net/http"
    "time"
)

func Index(w http.ResponseWriter, r *http.Request) {
    log.Println("receive a request from:", r.RemoteAddr, r.Header)
    time.Sleep(10 * time.Second)
    w.Write([]byte("ok"))
}

func main() {
    var s = http.Server{
        Addr:    ":8080",
        Handler: http.HandlerFunc(Index),
    }
    s.ListenAndServe()
}

我们看到这个http server的“不同之处”在于它不急于回复http应答，而是在接收请求10秒后再回复应答。下面是我们的http client端的代码：

// github.com/bigwhite/experiments/blob/master/http-client/default-client/client.go

package main

import (
    "fmt"
    "io"
    "net/http"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    wg.Add(256)
    for i := 0; i < 256; i++ {
        go func() {
            defer wg.Done()
            resp, err := http.Get("http://localhost:8080")
            if err != nil {
                panic(err)
            }
            defer resp.Body.Close()
            body, err := io.ReadAll(resp.Body)
            fmt.Println(string(body))
        }()
    }
    wg.Wait()
}

上面的客户端创建了256个goroutine，每个goroutine向server建立一条连接，我们先启动server，然后再运行一下上面的这个客户端程序：

$go run server.go
$$go run client.go
panic: Get "http://localhost:8080": dial tcp [::1]:8080: socket: too many open files

goroutine 25 [running]:
main.main.func1(0xc000128280)
    /Users/tonybai/Go/src/github.com/bigwhite/experiments/http-client/default-client/client.go:18 +0x1c7
created by main.main
    /Users/tonybai/Go/src/github.com/bigwhite/experiments/http-client/default-client/client.go:14 +0x78
exit status 2

我们看到上面的客户端抛出了一个panic，提示：打开文件描述符过多。

上面演示环境的ulimit -n的值为256

我们用一幅示意图来描述上面例子中的情况：

尽管根据《通过实例理解Go标准库http包是如何处理keep-alive连接的》一文我们知道，默认情况下，http客户端是会保持连接并复用到同一主机的服务的连接的。但由于上述示例中server的延迟10s回应答的上下文，客户端在默认情况下不会等待应答回来，而是尝试建立新的连接去发送新的http请求。由于示例运行环境最大允许每个进程打开256个文件描述符，因此在客户端后期向服务端建立连接时，就会出现“socket: too many open files”的错误。

2. 定义在小范围应用的http客户端实例

那么我们该如何控制客户端的行为以避免在资源受限的上下文情况下完成客户端的发送任务呢？我们通过设置http.DefaultClient的相关属性来实现这一点，但DefaultClient是包级变量，在整个程序中是共享的，一旦修改其属性，其他使用http默认客户端的包也会受到影响。因此更好的方案是定义一个在小范围应用的http客户端实例。

代码：

resp, err := http.Get("http://example.com/")
...
resp, err := http.Post("http://example.com/upload", "image/jpeg", &buf)
...

等价于如下代码：

client := &http.Client{} // 自定义一个http客户端实例
resp, err := client.Get("http://example.com/")
...
resp, err := client.Post("http://example.com/upload", "image/jpeg", &buf)
...

不同的是我们自定义的http.Client实例的应用范围仅限于上述特定范围，不会对其他使用http默认客户端的包产生任何影响。不过此时我们自定义的http.Client实例client的行为与DefaultClient的无异，要想解决上面示例panic的问题，我们还需对自定义的新客户端实例做一进步行为定制。

3. 定制到某一host的最大连接数

上述示例的最大问题在于向server端建立的连接数不受控制，即便将每个进程可以打开的最大文件描述符个数调大，客户端还可能会遇到最大向外建立的65535个连接的极限瓶颈(客户端socket端口用尽)，因此一个严谨的客户端需要设置到某个host的最大连接数限制。

那么，http.Client是如何控制到某个host的最大连接数的呢？http包的Client结构如下：

//$GOROOT/src/net/http/client.go

type Client struct {
        // Transport specifies the mechanism by which individual
        // HTTP requests are made.
        // If nil, DefaultTransport is used.
        Transport RoundTripper

    CheckRedirect func(req *Request, via []*Request) error
    Jar CookieJar
    Timeout time.Duration

Client结构体一共四个字段，能控制Client连接行为的是Transport字段。如果Transport的值为nil，那么Client的连接行为遵守DefaultTransport的设置：

// $GOROOT/src/net/http/transport.go

var DefaultTransport RoundTripper = &Transport{
        Proxy: ProxyFromEnvironment,
        DialContext: (&net.Dialer{
                Timeout:   30 * time.Second,
                KeepAlive: 30 * time.Second,
        }).DialContext,
        ForceAttemptHTTP2:     true,
        MaxIdleConns:          100,
        IdleConnTimeout:       90 * time.Second,
        TLSHandshakeTimeout:   10 * time.Second,
        ExpectContinueTimeout: 1 * time.Second,
}

不过在这份DefaultTransport的“配置”中，并没有有关向某个host建立最大连接数的设置，因为在Transport结构体中，起到这个作用的字段是MaxConnsPerHost：

// $GOROOT/src/net/http/transport.go

type Transport struct {
    ... ...

    // MaxConnsPerHost optionally limits the total number of
        // connections per host, including connections in the dialing,
        // active, and idle states. On limit violation, dials will block.
        //
        // Zero means no limit.
        MaxConnsPerHost int
    ... ...
}

我们来改造一下上面的示例：

// github.com/bigwhite/experiments/blob/master/http-client/client-with-maxconnsperhost/client.go

package main

import (
    "fmt"
    "io"
    "net/http"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    wg.Add(256)
    tr := &http.Transport{
        MaxConnsPerHost: 5,
    }
    client := http.Client{
        Transport: tr,
    }
    for i := 0; i < 256; i++ {
        go func(i int) {
            defer wg.Done()
            resp, err := client.Get("http://localhost:8080")
            if err != nil {
                panic(err)
            }
            defer resp.Body.Close()
            body, err := io.ReadAll(resp.Body)
            fmt.Printf("g-%d: %s\n", i, string(body))
        }(i)
    }
    wg.Wait()
}

上面的代码不再使用DefaultClient，而是自定义了一个新Client实例，并设置该实例的Transport字段为我们新建的设置了MaxConsPerHost字段的Transport实例。将server启动，并执行上面client.go，我们从server端看到如下结果：

$go run server.go

receive a request from: [::1]:63677 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63675 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63676 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63673 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63674 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:63673 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63675 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63674 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63676 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63677 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:63677 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63674 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63676 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63675 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:63673 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

我们看到：客户端一共向server端建立了5条连接(客户端端口号从63673到63677)，并且每隔10s，客户端复用这5条连接发送下一批请求。

http.Transport维护了到每个server host的计数器connsPerHost和请求等待队列：

// $GOROOT/src/net/http/transport.go
type Transport struct {
    ... ...
        connsPerHostMu   sync.Mutex
        connsPerHost     map[connectMethodKey]int
        connsPerHostWait map[connectMethodKey]wantConnQueue // waiting getConns
    ... ...
}

Transport结构体使用了一个connectMethodKey结构作为key：

// $GOROOT/src/net/http/transport.go
type connectMethodKey struct {
        proxy, scheme, addr string
        onlyH1              bool
}

我们看到connectMethodKey使用一个四元组(proxy,scheme,addr, onlyH1)来唯一标识一个“host”。通常对一个Client实例而言，proxy，scheme和onlyH1都是相同的，不同的是addr(ip+port)，因此实际上也就是按addr区分host。我们同样用一幅示意图描示意一下这种情况：

4. 设定idle池的大小

不知道大家是否想到这点：当上面示例中的到某一个host的五个链接没那么繁忙时，依旧保持这个五个链接是不是有些浪费资源呢？至少占用着客户端端口以及服务端的文件描述符资源。我们是否能让客户端在闲时减少保持的到服务端的链接数量呢？我们可以通过Transport结构体类型中的MaxIdleConnsPerHost字段实现这一点。

其实如果你不显式设置MaxIdleConnsPerHost，http包也会使用其默认值(2)：

// $GOROOT/src/net/http/transport.go

// DefaultMaxIdleConnsPerHost is the default value of Transport's
// MaxIdleConnsPerHost.
const DefaultMaxIdleConnsPerHost = 2

我们用一个例子来验证http.Client的这一行为！

首先我们改变一下server端的行为，将原先的“等待10s”改为立即返回应答：

// github.com/bigwhite/experiments/blob/master/http-client/client-with-maxidleconnsperhost/server.go

package main

import (
    "fmt"
    "net/http"
)

func Index(w http.ResponseWriter, r *http.Request) {
    fmt.Println("receive a request from:", r.RemoteAddr, r.Header)
    w.Write([]byte("ok"))
}

func main() {
    var s = http.Server{
        Addr:    ":8080",
        Handler: http.HandlerFunc(Index),
    }
    s.ListenAndServe()
}

而对于client，我们需要精心设计一下：

// github.com/bigwhite/experiments/blob/master/http-client/client-with-maxidleconnsperhost/client.go
     1  package main
     2
     3  import (
     4      "fmt"
     5      "io"
     6      "net/http"
     7      "sync"
     8      "time"
     9  )
    10
    11  func main() {
    12      var wg sync.WaitGroup
    13      wg.Add(5)
    14      tr := &http.Transport{
    15          MaxConnsPerHost:     5,
    16          MaxIdleConnsPerHost: 3,
    17      }
    18      client := http.Client{
    19          Transport: tr,
    20      }
    21      for i := 0; i < 5; i++ {
    22          go func(i int) {
    23              defer wg.Done()
    24              resp, err := client.Get("http://localhost:8080")
    25              if err != nil {
    26                  panic(err)
    27              }
    28              defer resp.Body.Close()
    29              body, err := io.ReadAll(resp.Body)
    30              fmt.Printf("g-%d: %s\n", i, string(body))
    31          }(i)
    32      }
    33      wg.Wait()
    34
    35      time.Sleep(10 * time.Second)
    36
    37      wg.Add(5)
    38      for i := 0; i < 5; i++ {
    39          go func(i int) {
    40              defer wg.Done()
    41
    42              for i := 0; i < 100; i++ {
    43                  resp, err := client.Get("http://localhost:8080")
    44                  if err != nil {
    45                      panic(err)
    46                  }
    47                  defer resp.Body.Close()
    48                  body, err := io.ReadAll(resp.Body)
    49                  fmt.Printf("g-%d: %s\n", i+10, string(body))
    50                  time.Sleep(time.Second)
    51              }
    52          }(i)
    53      }
    54      wg.Wait()
    55  }

我们首先制造一次忙碌的发送行为(21~32行)，使得client端建满5个连接；然后等待10s，即让client闲下来；之后再建立5个groutine，以每秒一条的速度向server端发送请求(不忙的节奏)，我们来看看程序运行后服务端的输出：

$go run server.go
receive a request from: [::1]:56244 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56246 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56242 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56243 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56245 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:56242 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56243 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56244 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56242 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56244 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:56243 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56242 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56244 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56243 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56244 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:56244 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56242 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56243 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56244 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56243 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:56244 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56242 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56243 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56244 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:56243 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

... ...

我们来分析一下：
- 第一部分的五行输出是“忙时”client端建立的5条不同的连接，客户端端口号从56242到56246；
- 第二部分的五行输出是“非忙时”client利用idle池中的连接发送的请求，关键点就在于这5个请求的源端口号：56242、56243和56244，五个请求使用了三个早已建立好的alive的连接；
- 后面的几部分使用的也是这三个早已建立好的alive的连接。

这就是MaxIdleConnsPerHost的作用：最初“忙时”建立的5条连接，在client进入闲时时要进入idle状态。但MaxIdleConnsPerHost的值为3，也就是说只有3条连接可以进入idle池，而另外两个会被close掉。于是源端口号为56242、56243和56244的三条连接被保留了下来。

下面是这节例子的示意图：

Transport结构体还有一个字段与idle池有关，那就是MaxIdleConns，不同于MaxIdleConnsPerHost只针对某个host，MaxIdleConns是针对整个Client的所有idle池中的连接数的和，这个和不能超过MaxIdleConns。

5. 清理idle池中的连接

如果没有其他设定，那么一个Client到一个host在闲时至少会保持DefaultMaxIdleConnsPerHost个idle连接(前提是之前已经建立了2条或2条以上的连接)，但如果Client针对这个host一直就保持无流量的状态，那么idle池中的连接也是一种资源浪费。于是Transport又提供了IdleConnTimeout字段用于超时清理idle池中的长连接。下面的示例复用上面的server，但client.go改为如下形式：

// github.com/bigwhite/experiments/blob/master/http-client/client-with-idleconntimeout/client.go

     1  package main
     2
     3  import (
     4      "fmt"
     5      "io"
     6      "net/http"
     7      "sync"
     8      "time"
     9  )
    10
    11  func main() {
    12      var wg sync.WaitGroup
    13      wg.Add(5)
    14      tr := &http.Transport{
    15          MaxConnsPerHost:     5,
    16          MaxIdleConnsPerHost: 3,
    17          IdleConnTimeout:     10 * time.Second,
    18      }
    19      client := http.Client{
    20          Transport: tr,
    21      }
    22      for i := 0; i < 5; i++ {
    23          go func(i int) {
    24              defer wg.Done()
    25              resp, err := client.Get("http://localhost:8080")
    26              if err != nil {
    27                  panic(err)
    28              }
    29              defer resp.Body.Close()
    30              body, err := io.ReadAll(resp.Body)
    31              fmt.Printf("g-%d: %s\n", i, string(body))
    32          }(i)
    33      }
    34      wg.Wait()
    35
    36      time.Sleep(5 * time.Second)
    37
    38      wg.Add(5)
    39      for i := 0; i < 5; i++ {
    40          go func(i int) {
    41              defer wg.Done()
    42              for i := 0; i < 2; i++ {
    43                  resp, err := client.Get("http://localhost:8080")
    44                  if err != nil {
    45                      panic(err)
    46                  }
    47                  defer resp.Body.Close()
    48                  body, err := io.ReadAll(resp.Body)
    49                  fmt.Printf("g-%d: %s\n", i+10, string(body))
    50                  time.Sleep(time.Second)
    51              }
    52          }(i)
    53      }
    54
    55      time.Sleep(15 * time.Second)
    56      wg.Add(5)
    57
    58      for i := 0; i < 5; i++ {
    59          go func(i int) {
    60              defer wg.Done()
    61              for i := 0; i < 100; i++ {
    62                  resp, err := client.Get("http://localhost:8080")
    63                  if err != nil {
    64                      panic(err)
    65                  }
    66                  defer resp.Body.Close()
    67                  body, err := io.ReadAll(resp.Body)
    68                  fmt.Printf("g-%d: %s\n", i+20, string(body))
    69                  time.Sleep(time.Second)
    70              }
    71          }(i)
    72      }
    73      wg.Wait()
    74  }

这个client.go代码分为三部分：首先和上个示例一样，我们首先制造一次忙碌的发送行为(22~33行)，使得client端建满5个连接；然后等待5s，即让client闲下来；之后再建立5个groutine，以每秒一条的速度向server端发送请求(不忙的节奏)；第三部分同样是先等待15s，然后创建5个goroutine分别以不忙的节奏向server端发送请求。我们来看看程序运行后服务端的输出：

$go run server.go

receive a request from: [::1]:52484 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52488 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52486 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52485 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52487 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:52487 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52488 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52484 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52484 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52487 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:52487 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52488 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52484 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52487 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52484 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:52542 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52544 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52545 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52543 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52546 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

receive a request from: [::1]:52542 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52544 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52545 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52542 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]
receive a request from: [::1]:52544 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]]

... ...

这里摘录5段输出。和预想的一样，第一段client向server建立了5条连接(客户端的端口号从52484~52487)；暂停5s后，新创建的5个goroutine通过idle池中的三条保持的连接向server发送请求(第二段和第三段，端口号:52484、52487、52488)；之后暂停15s，由于设置了IdleConnTimeout，idle池中的三条连接也被close掉了。这时再发送请求，client会重新建立连接（第四段，端口号52542~52546），最后一段则又开始通过idle池中的三条保持的连接向server发送请求了(端口号：52542、52544和52545)。

6. 其他控制项

如果觉得idle池超时清理依旧会占用“资源”一小会儿，那么可以利用Transport的DisableKeepAlives使得每个请求都创建一个新连接，即不复用keep-alive连接。当然这种控制设定在忙时导致的频繁建立新连接的损耗可是要比占用一些“资源”来的更大。示例可参考 github.com/bigwhite/experiments/blob/master/http-client/client-with-disablekeepalives，这里就不贴出来了。

另外像本文开始示例中server那样等待10s才回应答的行为可不是所有client端都能接受的，为了限定应答及时返回，client端可以设定等待应答的超时时间，如果超时，client将返回失败。http.Client结构中的Timeout可以实现这一特性。示例可参考 github.com/bigwhite/experiments/blob/master/http-client/client-with-timeout，这里同样不贴出来了。

本文涉及的代码可以在这里下载：https://github.com/bigwhite/experiments/blob/master/http-client。

“Gopher部落”知识星球正式转正（从试运营星球变成了正式星球）！“gopher部落”旨在打造一个精品Go学习和进阶社群！高品质首发Go技术文章，“三天”首发阅读权，每年两期Go语言发展现状分析，>每天提前1小时阅读到新鲜的Gopher日报，网课、技术专栏、图书内容前瞻，六小时内必答保证等满足你关于Go语言生态的所有需>求！部落目前虽小，但持续力很强。在2021年上半年，部落将策划两个专题系列分享，并且是部落独享哦：

Go技术书籍的书摘和读书体会系列
Go与eBPF系列

欢迎大家加入！

Go技术专栏“改善Go语⾔编程质量的50个有效实践”正在慕课网火热热销中！本专栏主要满足>广大gopher关于Go语言进阶的需求，围绕如何写出地道且高质量Go代码给出50条有效实践建议，上线后收到一致好评！欢迎大家订
阅！

img{512x368}

我的网课“Kubernetes实战：高可用集群搭建、配置、运维与应用”在慕课网热卖>中，欢迎小伙伴们订阅学习！

img{512x368}

我爱发短信：企业级短信平台定制开发专家 https://tonybai.com/。smspush : 可部署在企业内部的定制化短信平台，三网覆盖，不惧大并发接入，可定制扩展；短信内容你来定，不再受约束, 接口丰富，支持长短信，签名可选。2020年4月8日，中国三大电信运营商联合发布《5G消息白皮书》，51短信平台也会全新升级到“51商用消息平台”，全面支持5G RCS消息。

著名云主机服务厂商DigitalOcean发布最新的主机计划，入门级Droplet配置升级为：1 core CPU、1G内存、25G高速SSD，价格5$/月。有使用DigitalOcean需求的朋友，可以打开这个链接地址：https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式：

微博：https://weibo.com/bigwhite20xx
微信公众号：iamtonybai
博客：tonybai.com
github: https://github.com/bigwhite
“Gopher部落”知识星球：https://public.zsxq.com/groups/51284458844544

微信赞赏：
img{512x368}

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

使用Go实现可用select监听的队列

三月 26, 2021
0 条评论

1. 背景与选型

和《基于Redis Cluster的分布式锁实现以互斥方式操作共享资源》一文一样，今天要说的Go队列方案也是有一定项目背景的。

5G消息方兴未艾！前一段时间从事了一段时间5G消息网关的研发，但凡涉及类似消息业务的网关，我们一般都离不开队列这种数据结构的支持。这个5G消息网关项目采用的是Go技术栈开发，那么我们应该如何为它选择一个与业务模型匹配且性能不差的实现呢？

如今一提到消息队列，大家第一个想到的一定是kafka，kafka的确是一款优秀的分布式队列中间件，但对于我们这个系统来说，它有些“重”，部署和运维都有门槛，并且项目组里也没有能很好维护它的专家，毕竟“可控”是技术选择的一个重要因素。除此之外，我们更想在Go技术栈的生态中挑选，但kafka是Java实现的。

Go圈里在性能上能与kafka“掰掰手腕”的成熟选手不多，nats以及其主持持久化的子项目nats-streaming算是其中两个。不过nats的消息送达模型是：At-least-once-delivery，即至少送一次（而没有kafka的精确送一次的送达模型)。一旦消费者性能下降，给nats server返回的应答超时，nats就会做消息的重发处理：即将消息重新加入到队列中。这与我们的业务模型不符，即便nats提供了发送超时的设定，但我们还是无法给出适当的timeout时间。Go圈里的另一个高性能分布式消息队列nsq采用的也是“至少送一次”的消息送达模型，因此也无法满足我们的业务需求。

我们的业务决定了我们需要的队列要支持“多生产者多消费者”模型，Go语言内置的channel也是一个不错的候选。经过多个Go版本的打磨和优化，channel的send和recv操作性能在一定数量goroutine的情况下已经可以满足很多业务场景的需求了。但channel还是不完全满足我们的业务需求。我们的系统要求尽可能将来自客户端的消息接收下来并缓存在队列中。即便下游发送性能变慢，也要将客户消息先收下来，而不是拒收或延迟响应。而channel本质上是一个具有“静态大小”的队列并且Go的channel操作语义会在channel buffer满的情况下阻塞对channel的继续send，这就与我们的场景要求有背离，即便我们使用buffered channel，我们也很难选择一个合适的len值，并且一旦buffer满，它与unbuffered channel行为无异。

这样一来，我们便选择自己实现一个简单的、高性能的满足业务要求的队列，并且最好能像channel那样可以被select监听到数据ready，而不是给消费者带去“心智负担” ：消费者采用轮询的方式查看队列中是否有数据。

2. 设计与实现方案

要设计和实现这样一个队列结构，我们需要解决三个问题：

实现队列这个数据结构；
实现多goroutine并发访问队列时对消费者和生产者的协调；
解决消费者使用select监听队列的问题。

我们逐一来看！

1) 基础队列结构实现来自一个未被Go项目采纳的技术提案

队列是最基础的数据结构，实现一个“先进先出(FIFO)”的练手queue十分容易，但实现一份能加入标准库、资源占用小且性能良好的queue并不容易。Christian Petrin在2018年10月份曾发起一份关于Go标准库加入queue实现的技术提案，提案对基于array和链表的多种queue实现进行详细的比对，并最终给出结论：impl7是最为适宜和有竞争力的标准库queue的候选者。虽然该技术提案目前尚未得到accept，但impl7足可以作为我们的内存队列的基础实现。

2) 为impl7添加并发支持

在性能敏感的领域，我们可以直接使用sync包提供的诸多同步原语来实现goroutine并发安全访问，这里也不例外，一个最简单的让impl7队列实现支持并发的方法就是使用sync.Mutex实现对队列的互斥访问。由于impl7并未作为一个独立的repo存在，我们将其代码copy到我们的实现中(queueimpl7.go)，并将其包名由queueimpl7改名为queue：

// github.com/bigwhite/experiments/blob/master/queue-with-select/safe-queue1/queueimpl7.go

// Package queueimpl7 implements an unbounded, dynamically growing FIFO queue.
// Internally, queue store the values in fixed sized slices that are linked using
// a singly linked list.
// This implementation tests the queue performance when performing lazy creation of
// the internal slice as well as starting with a 1 sized slice, allowing it to grow
// up to 16 by using the builtin append function. Subsequent slices are created with
// 128 fixed size.
package queue

// Keeping below as var so it is possible to run the slice size bench tests with no coding changes.
var (
        // firstSliceSize holds the size of the first slice.
        firstSliceSize = 1

        // maxFirstSliceSize holds the maximum size of the first slice.
        maxFirstSliceSize = 16

        // maxInternalSliceSize holds the maximum size of each internal slice.
        maxInternalSliceSize = 128
)
... ...

下面我们就来为以queueimpl7为底层实现的queue增加并发访问支持：

// github.com/bigwhite/experiments/blob/master/queue-with-select/safe-queue1/safe-queue.go

package queue

import (
    "sync"
)

type SafeQueue struct {
    q *Queueimpl7
    sync.Mutex
}

func NewSafe() *SafeQueue {
    sq := &SafeQueue{
        q: New(),
    }

    return sq
}

func (s *SafeQueue) Len() int {
    s.Lock()
    n := s.q.Len()
    s.Unlock()
    return n
}

func (s *SafeQueue) Push(v interface{}) {
    s.Lock()
    defer s.Unlock()

    s.q.Push(v)
}

func (s *SafeQueue) Pop() (interface{}, bool) {
    s.Lock()
    defer s.Unlock()
    return s.q.Pop()
}

func (s *SafeQueue) Front() (interface{}, bool) {
    s.Lock()
    defer s.Unlock()
    return s.q.Front()
}

我们建立一个新结构体SafeQueue，用于表示支持并发访问的Queue，该结构只是在queueimpl7的Queue的基础上嵌入了sync.Mutex。

3) 支持select监听

到这里支持并发的queue虽然实现了，但在使用上还存在一些问题，尤其是对消费者而言，它只能通过轮询的方式来检查队列中是否有消息。而Go并发范式中，select扮演着重要角色，如果能让SafeQueue像普通channel那样能支持select监听，那么消费者在使用时的心智负担将大大降低。于是我们得到了下面第二版的SafeQueue实现：

// github.com/bigwhite/experiments/blob/master/queue-with-select/safe-queue2/safe-queue.go

package queue

import (
    "sync"
    "time"
)

const (
    signalInterval = 200
    signalChanSize = 10
)

type SafeQueue struct {
    q *Queueimpl7
    sync.Mutex
    C chan struct{}
}

func NewSafe() *SafeQueue {
    sq := &SafeQueue{
        q: New(),
        C: make(chan struct{}, signalChanSize),
    }

    go func() {
        ticker := time.NewTicker(time.Millisecond * signalInterval)
        defer ticker.Stop()
        for {
            select {
            case <-ticker.C:
                if sq.q.Len() > 0 {
                    // send signal to indicate there are message waiting to be handled
                    select {
                    case sq.C <- struct{}{}:
                        //signaled
                    default:
                        // not block this goroutine
                    }
                }
            }
        }

    }()

    return sq
}

func (s *SafeQueue) Len() int {
    s.Lock()
    n := s.q.Len()
    s.Unlock()
    return n
}

func (s *SafeQueue) Push(v interface{}) {
    s.Lock()
    defer s.Unlock()

    s.q.Push(v)
}

func (s *SafeQueue) Pop() (interface{}, bool) {
    s.Lock()
    defer s.Unlock()
    return s.q.Pop()
}

func (s *SafeQueue) Front() (interface{}, bool) {
    s.Lock()
    defer s.Unlock()
    return s.q.Front()
}

从上面代码看到，每个SafeQueue的实例会伴随一个goroutine，该goroutine会定期(signalInterval)扫描其所绑定的队列实例中当前消息数，如果大于0，则会向SafeQueue结构中新增的channel发送一条数据，作为一个“事件”。SafeQueue的消费者则可以通过select来监听该channel，待收到“事件”后调用SafeQueue的Pop方法获取队列数据。下面是一个SafeQueue的简单使用示例：

// github.com/bigwhite/experiments/blob/master/queue-with-select/main.go
package main

import (
    "fmt"
    "sync"
    "time"

    queue "github.com/bigwhite/safe-queue/safe-queue2"
)

func main() {
    var q = queue.NewSafe()
    var wg sync.WaitGroup

    wg.Add(2)
    // 生产者
    go func() {
        for i := 0; i < 1000; i++ {
            time.Sleep(time.Second)
            q.Push(i + 1)

        }
        wg.Done()
    }()

    // 消费者
    go func() {
    LOOP:
        for {
            select {
            case <-q.C:
                for {
                    i, ok := q.Pop()
                    if !ok {
                        // no msg available
                        continue LOOP
                    }

                    fmt.Printf("%d\n", i.(int))
                }
            }

        }

    }()

    wg.Wait()
}

从支持SafeQueue的原理可以看到，当有多个消费者时，只有一个消费者能得到“事件”并开始消费。如果队列消息较少，只有一个消费者可以启动消费，这个机制也不会导致“惊群”；当队列中有源源不断的消费产生时，与SafeQueue绑定的goroutine可能会连续发送“事件”，多个消费者都会收到事件并启动消费行为。在这样的实现下，建议消费者在收到“事件”后持续消费，直到Pop的第二个返回值返回false(代表队列为空)，就像上面示例中的那样。

这个SafeQueue的性能“中规中矩”，比buffered channel略好(Go 1.16 darwin下跑的benchmark)：

$go test -bench .
goos: darwin
goarch: amd64
pkg: github.com/bigwhite/safe-queue/safe-queue2
cpu: Intel(R) Core(TM) i5-8257U CPU @ 1.40GHz
BenchmarkParallelQueuePush-8                10687545           110.9 ns/op        32 B/op          1 allocs/op
BenchmarkParallelQueuePop-8                 18185744            55.58 ns/op        0 B/op          0 allocs/op
BenchmarkParallelPushBufferredChan-8        10275184           127.1 ns/op        16 B/op          1 allocs/op
BenchmarkParallelPopBufferedChan-8          10168750           128.8 ns/op        16 B/op          1 allocs/op
BenchmarkParallelPushUnBufferredChan-8       3005150           414.9 ns/op        16 B/op          1 allocs/op
BenchmarkParallelPopUnBufferedChan-8         2987301           402.9 ns/op        16 B/op          1 allocs/op
PASS
ok      github.com/bigwhite/safe-queue/safe-queue2  11.209s

注：BenchmarkParallelQueuePop-8因为是读取空队列，所以没有分配内存，实际情况是会有内存分配的。另外并发goroutine的模拟差异可能导致有结果差异。

3. 扩展与问题

上面实现的SafeQueue是一个纯内存队列，一旦程序停止/重启，未处理的消息都将消失。一个传统的解决方法是采用wal(write ahead log)在推队列之前将消息持久化后写入文件，在消息出队列后将消息状态也写入wal文件中。这样重启程序时，从wal中恢复消息到各个队列即可。我们也可以将wal封装到SafeQueue的实现中，在SafeQueue的Push和Pop时自动操作wal，并对SafeQueue的使用者透明，不过这里有一个前提，那就是队列消息的可序列化（比如使用protobuf)。另外SafeQueue还需提供一个对外的wal消息恢复接口。大家可以考虑一下如何实现这些。

另外在上述的SafeQueue实现中，我们在给SafeQueue增加select监听时引入两个const：

const (
    signalInterval = 200
    signalChanSize = 10
)

对于SafeQueue的使用者而言，这两个默认值可能不满足需求，那么我们可以将SafeQueue的New方法做一些改造，采用“功能选项(functional option)”的模式为用户提供设置这两个值的可选接口，这个“作业”也留给大家了^_^。

本文所有示例代码可以在这里下载 – https://github.com/bigwhite/experiments/tree/master/queue-with-select。

Go技术书籍的书摘和读书体会系列
Go与eBPF系列

欢迎大家加入！

img{512x368}

我的网课“Kubernetes实战：高可用集群搭建、配置、运维与应用”在慕课网热卖>中，欢迎小伙伴们订阅学习！

img{512x368}

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式：

微博：https://weibo.com/bigwhite20xx
微信公众号：iamtonybai
博客：tonybai.com
github: https://github.com/bigwhite
“Gopher部落”知识星球：https://public.zsxq.com/groups/51284458844544

微信赞赏：
img{512x368}

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

标签标准库下的文章

http.Client的连接行为控制详解

1. http包默认客户端

2. 定义在小范围应用的http客户端实例

3. 定制到某一host的最大连接数

4. 设定idle池的大小

5. 清理idle池中的连接

6. 其他控制项

使用Go实现可用select监听的队列

1. 背景与选型

2. 设计与实现方案

1) 基础队列结构实现来自一个未被Go项目采纳的技术提案

2) 为impl7添加并发支持

3) 支持select监听

3. 扩展与问题

欢迎使用邮件订阅我的博客

文章

评论

分类

归档

链接

开源项目

翻译项目

标签 标准库 下的文章

http.Client的连接行为控制详解

1. http包默认客户端

2. 定义在小范围应用的http客户端实例

3. 定制到某一host的最大连接数

4. 设定idle池的大小

5. 清理idle池中的连接

6. 其他控制项

使用Go实现可用select监听的队列

1. 背景与选型

2. 设计与实现方案

1) 基础队列结构实现来自一个未被Go项目采纳的技术提案

2) 为impl7添加并发支持

3) 支持select监听

3. 扩展与问题

欢迎使用邮件订阅我的博客

文章

评论

分类

标签

归档

链接

开源项目

翻译项目

标签标准库下的文章