为什么response.Body.Close()是必要的

最近用go写web时，在网上简单查了下客户端请求的代码，发现有的在读取response后会执行response.Body.Close()，有的则不会。在网上查阅了下发现如果不执行可能会导致goroutine泄漏。本着质疑到底的精神，笔者今天就做个实验验证一下，并对相关源码作一下深入的解析

受篇幅和冗余细节的限制，笔者将着重聚焦于整个流程和函数功能的阐释，而不会过分关注无关代码

验证

先来简单做个小验证

不执行response.Body.Close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
package main

import (
	"fmt"
	"net/http"
	"os"
)

func main() {
	println("pid:", os.Getpid())
	for true {
		requestWithClose()
	}
}

func requestWithClose() {
	resp, err := http.Get("https://www.baidu.com")
	if err != nil {
		fmt.Printf("error occurred while fetching page, error: %s", err.Error())
		return
	}
	defer resp.Body.Close()
	fmt.Println("ok")
}
执行response.Body.Close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
package main

import (
	"fmt"
	"net/http"
	"os"
)

func main() {
	println("pid:", os.Getpid())
	for true {
		requestWithNoClose()
	}
}

func requestWithNoClose() {
	_, err := http.Get("https://www.baidu.com")
	if err != nil {
		fmt.Printf("error occurred while fetching page, error: %s", err.Error())
	}
	fmt.Println("ok")
}

在程序运行一段时间后，对对应进程执行top指令查看内存情况时，会发现如下结果：

pic1

可以发现如果不执行response.Body.Close()，整个进程所消耗的内存资源是会不断增加的

流程分析

引入

让我们从最开始讲起……

1
2
3
4
5
6
7
8
func main() {
    //调用get方法请求获得response
    resp, err := http.Get("https://www.baidu.com")
    if err != nil {
        panic(err)
    }
    resp.Body.Close()
}

http的get/post等请求最终会调用client.go下的func (c *Client) do(req *Request) (retres *Response, reterr error)这个方法。client就是请求的客户端，一般是采用默认的DefaultClient来作为方法的接收者进行调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
func (c *Client) do(req *Request) (retres *Response, reterr error) {
	//与流程相关的核心代码如下：
    
    //……
    
	for {
        
    	//……
        
        reqs = append(reqs, req)
        var err error
        var didTimeout func() bool
    	//调用client.send方法来获取response
        if resp, didTimeout, err = c.send(req, deadline); err != nil {
            // c.send() always closes req.Body
            reqBodyClosed = true
            if !deadline.IsZero() && didTimeout() {
                err = &httpError{
                    // TODO: early in cycle: s/Client.Timeout exceeded/timeout or context cancelation/
                    err:     err.Error() + " (Client.Timeout exceeded while awaiting headers)",
                    timeout: true,
                }
            }
            return nil, uerr(err)
        }
        
    	//……
        
    }
}

可以看到，关键的代码实际上就在c.send(req, deadline)这一行，那我们继续进入这个方法中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
func (c *Client) send(req *Request, deadline time.Time) (resp *Response, didTimeout func() bool, err error) {
    if c.Jar != nil {
        for _, cookie := range c.Jar.Cookies(req.URL) {
            req.AddCookie(cookie)
        }
    }
  //调用send方法来获取response
    resp, didTimeout, err = send(req, c.transport(), deadline)
    if err != nil {
        return nil, didTimeout, err
    }
    if c.Jar != nil {
        if rc := resp.Cookies(); len(rc) > 0 {
            c.Jar.SetCookies(req.URL, rc)
        }
    }
    return resp, nil, nil
}

接着进入send(req, c.transport(), deadline)方法中

1
2
3
4
5
6
7
8
9
func send(ireq *Request, rt RoundTripper, deadline time.Time) (resp *Response, didTimeout func() bool, err error) {
    
	//……
    //调用Transport.RoundTrip来处理请求
    resp, err = rt.RoundTrip(req)

	//……
    
}

在进入rt.RoundTrip(req)方法前，我们要先看一下这个方法的接收者——Transport

Transport

transport实现了RoundTripper接口，该接口只有一个方法RoundTrip()，故transport的入口函数就是RoundTrip()。transport的主要功能其实就是缓存了长连接，用于大量http请求场景下的连接复用，减少发送请求时TCP(TLS)连接建立的时间损耗，同时transport还能对连接做一些限制，如连接超时时间，每个host的最大连接数等

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
type Transport struct {
    idleMu     sync.Mutex
    wantIdle   bool                                // user has requested to close all idle conns
    
    //缓存空闲连接的地方
    idleConn   map[connectMethodKey][]*persistConn // most recently used at end
    
    //connectMethodKey => 空闲连接的chan形成的map
    //有空闲连接放入的时候，首先尝试放入这个chan，方便另一个可能需要连接的goroutine直接使用，如果没有goroutine需要连接，就放入到上面的idleConn里面，便于后面的请求连接复用
    idleConnCh map[connectMethodKey]chan *persistConn
    
    // DisableKeepAlives, if true, disables HTTP keep-alives and
    // will only use the connection to the server for a single
    // HTTP request.
    //
    // This is unrelated to the similarly named TCP keep-alives.
    //是否开启keepAlive，为true的话，连接不会被复用
    //设成Disable的目的个人猜测应该是因为bool类型默认为false，表示默认开启连接复用
    DisableKeepAlives bool
    
    // MaxIdleConns controls the maximum number of idle (keep-alive)
    // connections across all hosts. Zero means no limit.
    //所有hosts对应的最大的连接总数
    MaxIdleConns int
    
   //每一个host对应的最大的空闲连接数
    MaxIdleConnsPerHost int
    
    //每一个host对应的最大连接数
    MaxConnsPerHost int
}

Transport.roundTrip

接下来就是重点了

rt.RoundTrip(req)这个方法的主要目的是：

  1. 参数校验: scheme, host, method, protocol…
  2. 获取缓存的或新建的连接
  3. 从获取的连接中取得response
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
func (t *Transport) roundTrip(req *Request) (*Response, error) {
	
    //……
    
	for {
		select {
		case <-ctx.Done():
			req.closeBody()
			return nil, ctx.Err()
		default:
		}

		// treq gets modified by roundTrip, so we need to recreate for each retry.
		treq := &transportRequest{Request: req, trace: trace, cancelKey: cancelKey}
		cm, err := t.connectMethodForRequest(treq)
		if err != nil {
			req.closeBody()
			return nil, err
		}

		// Get the cached or newly-created connection to either the
		// host (for http or https), the http proxy, or the http proxy
		// pre-CONNECTed to https server. In any case, we'll be ready
		// to send it requests.
        //这里是关键，根据请求和connectMethod建立一个可用的连接（缓存/新建）
		pconn, err := t.getConn(treq, cm)
		if err != nil {
			t.setReqCanceler(cancelKey, nil)
			req.closeBody()
			return nil, err
		}

		var resp *Response
		if pconn.alt != nil {
			// HTTP/2 path.
			t.setReqCanceler(cancelKey, nil) // not cancelable with CancelRequest
			resp, err = pconn.alt.RoundTrip(req)
		} else {
            //这里也是关键，获取response
			resp, err = pconn.roundTrip(treq)
		}
		if err == nil {
			resp.Request = origReq
			return resp, nil
		}

		//……
        
	}
}

这个方法的核心在两个地方，一个是通过t.getConn(treq, cm)获得可用的persistConn连接；另一个是调用pconn.roundTrip(treq)方法获取到response。让我们相继解释一下两个方法

Transport.getConn

这个方法分成以下几个步骤：

  1. 调用t.queueForIdleConn获取一个空闲且可复用的连接，如果获取成功则直接返回该连接
  2. 如果未获取到空闲连接则调用t.queueForDial开始新建一个连接
  3. 等待w.ready关闭，则可以返回新的连接
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
func (t *Transport) getConn(treq *transportRequest, cm connectMethod) (pc *persistConn, err error) {
    req := treq.Request
    trace := treq.trace
    ctx := req.Context()
    
    //……
    
    w := &wantConn{
        cm:         cm,
        key:        cm.key(),
        ctx:        ctx,
        ready:      make(chan struct{}, 1),
        beforeDial: testHookPrePendingDial,
        afterDial:  testHookPostPendingDial,
    }
    
	//……
    
	// Queue for idle connection.
    //从idleConn里面获取一个connectMethod对应的空闲的连接
	if delivered := t.queueForIdleConn(w); delivered {
		pc := w.pc
        
		//……
        
		return pc, nil
	}

	cancelc := make(chan error, 1)
	t.setReqCanceler(treq.cancelKey, func(err error) { cancelc <- err })

    
    
	// Queue for permission to dial.
    //这个方法用来新建连接
	t.queueForDial(w)

	// Wait for completion or cancellation.
	select {
	case <-w.ready:
		// Trace success but only for HTTP/1.
		// HTTP/2 calls trace.GotConn itself.
		if w.pc != nil && w.pc.alt == nil && trace != nil && trace.GotConn != nil {
			trace.GotConn(httptrace.GotConnInfo{Conn: w.pc.conn, Reused: w.pc.isReused()})
		}
        
		//……
		
        return w.pc, w.err
	case <-req.Cancel:
		return nil, errRequestCanceledConn
	case <-req.Context().Done():
		return nil, req.Context().Err()
	case err := <-cancelc:
		if err == errRequestCanceled {
			err = errRequestCanceledConn
		}
		return nil, err
	}
}

这个方法又会导向以下两个函数：

Transport.queueForIdleConn

(*Transport).queueForIdleConn方法会根据请求的connectMethodKey从t.idleConn获取一个[]*persistConn切片， 并从切片中，根据算法获取一个有效的空闲连接。如果未获取到空闲连接，则将wantConn结构体变量（即形参w）放入t.idleConnWait[w.key]等待队列

Transport.queueForDial

这个方法比较复杂，主要涵盖获得一个新的连接及将其放入连接池的逻辑。因为很多地方与本文内容关系不大，暂且略过一些无关紧要之处，以后有机会再详谈。让我们把目光集中到它最终导向的一个很重要的方法——Transport.dialConn上

Transport.dialConn

这个方法的主要逻辑如下：

  1. 调用t.dial(ctx, "tcp", cm.addr())创建TCP连接
  2. 如果是https的请求， 则对请求建立安全的tls传输通道
  3. 为persistConn创建读写buffer， 如果用户没有自定义读写buffer的大小， 根据writeBufferSize和readBufferSize方法可知， 读写bufffer的大小默认为4096
  4. 执行go pconn.readLoop()和go pconn.writeLoop()开启读写循环然后返回连接
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
func (t *Transport) dialConn(ctx context.Context, cm connectMethod) (pconn *persistConn, err error) {
	pconn = &persistConn{
		t:             t,
		cacheKey:      cm.key(),
		reqch:         make(chan requestAndChan, 1),
		writech:       make(chan writeRequest, 1),
		closech:       make(chan struct{}),
		writeErrCh:    make(chan error, 1),
		writeLoopDone: make(chan struct{}),
	}
    
	//……
    
	//与tls有关的一些逻辑
    if cm.scheme() == "https" && t.hasCustomTLSDialer() {
		
    	//……
        
    //创建tcp连接
	} else {
		conn, err := t.dial(ctx, "tcp", cm.addr())
		if err != nil {
			return nil, wrapErr(err)
		}
		pconn.conn = conn
		if cm.scheme() == "https" {
			var firstTLSHost string
			if firstTLSHost, _, err = net.SplitHostPort(cm.addr()); err != nil {
				return nil, wrapErr(err)
			}
			if err = pconn.addTLS(firstTLSHost, trace); err != nil {
				return nil, wrapErr(err)
			}
		}
	}

	//……（有关proxy代理的一些逻辑）
    
	//创建读写buffer
	pconn.br = bufio.NewReaderSize(pconn, t.readBufferSize())
	pconn.bw = bufio.NewWriterSize(persistConnWriter{pconn}, t.writeBufferSize())
    
	//开启读写循环
	go pconn.readLoop()
	go pconn.writeLoop()
	return pconn, nil
}

//创建读写buffer的两个方法
func (t *Transport) writeBufferSize() int {
    if t.WriteBufferSize > 0 {
        return t.WriteBufferSize
    }
    return 4 << 10
}

func (t *Transport) readBufferSize() int {
    if t.ReadBufferSize > 0 {
        return t.ReadBufferSize
    }
    return 4 << 10
}

接下来就到了本文最关键的两个方法：读写循环了。但是先等一等，别忘了还有另一条线的逻辑我们搁置了好久，让我们把两条线合在一起

persistConn.roundTrip

persistConn是一个具体的连接实例，包括连接的上下文。而它的persistConn.roundTrip方法的目的其实就是获取到response并返回给上层，步骤如下：

  1. 向连接的writech写入writeRequest: pc.writech <- writeRequest{req, writeErrCh, continueCh}, 参考(*Transport).dialConn可知pc.writech是一个缓冲大小为1的管道，所以会立即写入成功。writech由写循环读
  2. 向连接的reqch写入requestAndChan: pc.reqch <- requestAndChan, pc.reqch和pc.writech一样都是缓冲大小为1的管道。reqch用于获取客户端的请求信息并等待返回的response，由读循环读
  3. 开启for循环select， 等待响应或者超时等信息

归根结底，这个函数的目的其实也就是创建个writeLoop 用到的chan，然后把request通过这个chan传给 persistConn.writeLoop ，然后再创建一个readLoop用到的chan，从这个chan中获取 persistConn.readLoop 获取到的 response，最后把response返回给上层函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err error) {
	
    //……

	gone := make(chan struct{})
	defer close(gone)

	//……

	const debugRoundTrip = false

	// Write the request concurrently with waiting for a response,
	// in case the server decides to reply before reading our full
	// request body.
	startBytesWritten := pc.nwrite
	writeErrCh := make(chan error, 1)
    //向通道写入request
	pc.writech <- writeRequest{req, writeErrCh, continueCh}

	resc := make(chan responseAndError)
	pc.reqch <- requestAndChan{
		req:        req.Request,
		cancelKey:  req.cancelKey,
		ch:         resc,
		addedGzip:  requestedGzip,
		continueCh: continueCh,
		callerGone: gone,
	}

	var respHeaderTimer <-chan time.Time
	cancelChan := req.Request.Cancel
	ctxDoneChan := req.Context().Done()
	pcClosed := pc.closech
	canceled := false
	for {
		testHookWaitResLoop()
		select {
		case err := <-writeErrCh:
			if debugRoundTrip {
				req.logf("writeErrCh resv: %T/%#v", err, err)
			}
			if err != nil {
				pc.close(fmt.Errorf("write error: %v", err))
				return nil, pc.mapRoundTripError(req, startBytesWritten, err)
			}
			if d := pc.t.ResponseHeaderTimeout; d > 0 {
				if debugRoundTrip {
					req.logf("starting timer for %v", d)
				}
				timer := time.NewTimer(d)
				defer timer.Stop() // prevent leaks
				respHeaderTimer = timer.C
			}
		case <-pcClosed:
			pcClosed = nil
			if canceled || pc.t.replaceReqCanceler(req.cancelKey, nil) {
				if debugRoundTrip {
					req.logf("closech recv: %T %#v", pc.closed, pc.closed)
				}
				return nil, pc.mapRoundTripError(req, startBytesWritten, pc.closed)
			}
		case <-respHeaderTimer:
			if debugRoundTrip {
				req.logf("timeout waiting for response headers.")
			}
			pc.close(errTimeout)
			return nil, errTimeout
		case re := <-resc:
			if (re.res == nil) == (re.err == nil) {
				panic(fmt.Sprintf("internal error: exactly one of res or err should be set; nil=%v", re.res == nil))
			}
			if debugRoundTrip {
				req.logf("resc recv: %p, %T/%#v", re.res, re.err, re.err)
			}
			if re.err != nil {
				return nil, pc.mapRoundTripError(req, startBytesWritten, re.err)
			}
			return re.res, nil
		case <-cancelChan:
			canceled = pc.t.cancelRequest(req.cancelKey, errRequestCanceled)
			cancelChan = nil
		case <-ctxDoneChan:
			canceled = pc.t.cancelRequest(req.cancelKey, req.Context().Err())
			cancelChan = nil
			ctxDoneChan = nil
		}
	}
}

persistConn.readLoop&persistConn.writeLoop

终于到了引起内存泄漏的两个函数了，先看简单的写循环：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
func (pc *persistConn) writeLoop() {
	defer close(pc.writeLoopDone)
	for {
		select {
        //获取到用户写入的请求
		case wr := <-pc.writech:
			startBytesWritten := pc.nwrite
            //将用户的请求写入连接的写缓存buffer
			err := wr.req.Request.write(pc.bw, pc.isProxy, wr.req.extra, pc.waitForContinue(wr.continueCh))
			if bre, ok := err.(requestBodyReadError); ok {
				err = bre.error
				// Errors reading from the user's
				// Request.Body are high priority.
				// Set it here before sending on the
				// channels below or calling
				// pc.close() which tears town
				// connections and causes other
				// errors.
				wr.req.setError(err)
			}
            //清除缓存，发送请求
			if err == nil {
				err = pc.bw.Flush()
			}
			if err != nil {
				wr.req.Request.closeBody()
				if pc.nwrite == startBytesWritten {
					err = nothingWrittenError{err}
				}
			}
			pc.writeErrCh <- err // to the body reader, which might recycle us
			wr.ch <- err         // to the roundTrip function
			if err != nil {
				pc.close(err)
				return
			}
		case <-pc.closech:
			return
		}
	}
}

其实就是把用户的请求写入连接的写缓存buffer，最后再flush

接下来是读循环

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
func (pc *persistConn) readLoop() {
    //结束时关闭连接，同时也会关闭pc.closech chan，因此writeLoop也不会阻塞
	closeErr := errReadLoopExiting // default value, if not changed below
	defer func() {
		pc.close(closeErr)
		pc.t.removeIdleConn(pc)
	}()

	//……
    
	alive := true
	for alive {
		
        //……
        
		//从连接中读request
		rc := <-pc.reqch
		trace := httptrace.ContextClientTrace(rc.req.Context())
		//从连接的读buffer中获取response
		var resp *Response
		if err == nil {
			resp, err = pc.readResponse(rc, trace)
		} else {
			err = transportReadFromServerError{err}
			closeErr = err
		}
        
        //……（请求类型为HEAD/body为空的处理）
        
		//这个通道用于等待response被读取
		waitForBodyRead := make(chan bool, 2)
		body := &bodyEOFSignal{
			body: resp.Body,
            //这个方法是resp.Body.Close()最终调用的函数，向waitForBodyRead通道写入false
			earlyCloseFn: func() error {
				waitForBodyRead <- false
				<-eofc // will be closed by deferred call at the end of the function
				return nil

			},
            //判断response是否读完， 如果已经读完则向waitForBodyRead写入true否则写入false
			fn: func(err error) error {
                isEOF := err == io.EOF
                waitForBodyRead <- isEOF
                if isEOF {
                    <-eofc // see comment above eofc declaration
                } else if err != nil {
                    if cerr := pc.canceled(); cerr != nil {
                        return cerr
                    }
                }
                return err
            },
		}

        //组装response
		resp.Body = body
		if rc.addedGzip && strings.EqualFold(resp.Header.Get("Content-Encoding"), "gzip") {
			resp.Body = &gzipReader{body: body}
			resp.Header.Del("Content-Encoding")
			resp.Header.Del("Content-Length")
			resp.ContentLength = -1
			resp.Uncompressed = true
		}
		
        //将response传递给上层函数（即persistConn.roundTrip），通过rc.ch（该通道用于存储响应）
		select {
		case rc.ch <- responseAndError{res: resp}:
		case <-rc.callerGone:
			return
		}

        //函数最终阻塞在这里判断下一步
		// Before looping back to the top of this function and peeking on
		// the bufio.Reader, wait for the caller goroutine to finish
		// reading the response body. (or for cancellation or death)
		select {
		case bodyEOF := <-waitForBodyRead:
			replaced := pc.t.replaceReqCanceler(rc.cancelKey, nil) // before pc might return to idle pool
			alive = alive &&
				bodyEOF &&
				!pc.sawEOF &&
				pc.wroteRequest() &&
				replaced && tryPutIdleConn(trace)
			if bodyEOF {
				eofc <- struct{}{}
			}
		case <-rc.req.Cancel:
			alive = false
			pc.t.CancelRequest(rc.req)
		case <-rc.req.Context().Done():
			alive = false
			pc.t.cancelRequest(rc.cancelKey, rc.req.Context().Err())
		case <-pc.closech:
			alive = false
		}

		testHookReadLoopBeforeNextRead()
	}
}

解析到这里，大家应该也发现关键所在了：由于readLoop和writeLoop两个goroutine在写入请求并获取response返回后，并没有跳出for循环，而是继续阻塞在下一次for循环的select语句里面，所以，两个函数所在的goroutine并没有运行结束，自然也就导致了goroutine的泄露

为什么response.Body.Close()可以结束阻塞

其实close的主要逻辑就是通过调用 readLoop 定义的earlyCloseFn 方法，向waitForBodyRead的chan写入false，进而让readLoop退出阻塞，继而终止readLoop的goroutine。而readLoop在退出的时候，会关闭closech chan，进而让 writeLoop退出阻塞，终止writeLoop的goroutine

总结

由于底层的读写循环会一直阻塞，直到我们对Body进行Close或者把当前http请求的body读完。所以如果不对response.Body进行close，goroutine就会一直阻塞在原先的位置，连接也不能复用。感兴趣再研究一下代码会发现：

  • 有Body没有读完，就执行Close的话，连接会被关闭，不会复用
  • 如果读完了所有Body，可以不调用Close，连接也会被复用

所以最佳实践是：http请求发出后，如果没有错误，马上声明defer res.Body.Close()，避免资源泄露，然后尽量读完Body，好让连接复用

最后依旧用一张图来描述下整个流程：

pic1

参考

Go Http包解析：为什么需要response.Body.Close() - SegmentFault 思否

Go中的HTTP请求之——HTTP1.1请求流程分析 - SegmentFault 思否

赏

谢谢你请我吃糖果

扫一扫，分享到微信

微信分享二维码