会开启 HTTP 事务发送 HTTP 请求并等待远程的响应，经过下面一连串的调用，我们最终来到了标准库实现底层 HTTP 协议的结构体 — net/http.Transport：
net/http.Client.Donet/http.Client.donet/http.Client.sendnet/http.sendnet/http.Transport.RoundTrip- 、
在创建了Request结构体req,指定了请求的Method和url。
- 会调用send方法来发送客户端请求，并获取服务端的响应信息。
- 如果服务端回复的不需要重定向，则将该响应resp返回
- 如果服务端回复的需要重定向，则获取重定向的Request，并进行重定向校验
- 重定向校验通过后，会继续调用send方法来发送重定向的请求。
- 不需要重定向时返回从服务端响应的结果resp。
send方法
func (c *Client) send(req *Request, deadline time.Time) (resp *Response, didTimeout func() bool, err error) {
//如果Jar不为nil，则将Jar中的Cookie添加到请求中
if c.Jar != nil {
for _, cookie := range c.Jar.Cookies(req.URL) {
req.AddCookie(cookie)
}
}
//发送请求到服务端，并返回从服务端读取到的response信息resp
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方法主要调用send函数将请求发送到Transport中，并返回response。
如果Jar不为nil，处理 HTTP 响应中的 cookies 并将其存储在 CookieJar 中，以便后续请求使用，并返回从服务端读取到的response信息resp。需要注意的是如果用户自定义了Transport则用用户自定义的，如果没有则用默认的DefaultTransport。
send issues an HTTP request.
该函数的主要作用是调用RoundTrip完成一个HTTP事务，并返回一个resp。
// send issues an HTTP request.
// Caller should close resp.Body when done reading from it.
func send(ireq *Request, rt RoundTripper, deadline time.Time) (resp *Response, didTimeout func() bool, err error) {
req := ireq // req is either the original request, or a modified fork
/*
条件判断：
如果RoundTripper为nil 或者请求的URL为nil 或者 请求的RequestURI不为空，则关闭请求体，返回error
RequestURI 是给server端使用的
*/
// forkReq forks req into a shallow clone of ireq the first
// time it's called.第一次调用时，将req转换为ireq的拷贝
forkReq := func() {
if ireq == req {
req = new(Request)
*req = *ireq // shallow clone
}
}
// Most the callers of send (Get, Post, et al) don't need
// Headers, leaving it uninitialized. We guarantee to the
// Transport that this has been initialized, though.
/*
由于大多数的调用(Get,Post等)都不需要Headers，而是将其保留成为初始化状态。不过我们传输到Transport需要保证其被初始化，所以这里将
没有Header为nil的进行初始化
如果请求头为nil
*/
if req.Header == nil {
forkReq()
req.Header = make(Header)
}
/*
如果URL中协议用户和密码信息，并且请求头的Authorization为空，我们需要设置Header的Authorization
*/
if u := req.URL.User; u != nil && req.Header.Get("Authorization") == "" {
username := u.Username()
password, _ := u.Password()
forkReq()
req.Header = cloneOrMakeHeader(ireq.Header)
req.Header.Set("Authorization", "Basic "+basicAuth(username, password))
}
//如果设置了超时时间，则需要调用forkReq，来确保req是ireq的拷贝，而不是执行同一地址的指针
if !deadline.IsZero() {
forkReq()
}
//根据deadline设置超时
stopTimer, didTimeout := setRequestCancel(req, rt, deadline)
//调用RoundTrip完成一个HTTP事务，并返回一个resp
resp, err = rt.RoundTrip(req)
if err != nil {
stopTimer() //取消监听超时
if resp != nil {
log.Printf("RoundTripper returned a response & error; ignoring response")
}
if tlsErr, ok := err.(tls.RecordHeaderError); ok {
// If we get a bad TLS record header, check to see if the
// response looks like HTTP and give a more helpful error.
// See golang.org/issue/11111.
if string(tlsErr.RecordHeader[:]) == "HTTP/" {
err = ErrSchemeMismatch
}
}
return nil, didTimeout, err
}
if resp == nil {
return nil, didTimeout, fmt.Errorf("http: RoundTripper implementation (%T) returned a nil *Response with a nil error", rt)
}
if resp.Body == nil {
// The documentation on the Body field says “The http Client and Transport
// guarantee that Body is always non-nil, even on responses without a body
// or responses with a zero-length body.” Unfortunately, we didn't document
// that same constraint for arbitrary RoundTripper implementations, and
// RoundTripper implementations in the wild (mostly in tests) assume that
// they can use a nil Body to mean an empty one (similar to Request.Body).
// (See https://golang.org/issue/38095.)
//
// If the ContentLength allows the Body to be empty, fill in an empty one
// here to ensure that it is non-nil.
if resp.ContentLength > 0 && req.Method != "HEAD" {
return nil, didTimeout, fmt.Errorf("http: RoundTripper implementation (%T) returned a *Response with content length %d but a nil Body", rt, resp.ContentLength)
}
resp.Body = io.NopCloser(strings.NewReader(""))
}
if !deadline.IsZero() { //如果设置了超时，则将Body转成cancelTimerBody
resp.Body = &cancelTimerBody{
stop: stopTimer,
rc: resp.Body,
reqDidTimeout: didTimeout,
}
}
return resp, nil, nil
}
对于 stopTimer, didTimeout := setRequestCancel(req, rt, deadline) 分析延后
看下resp, err = rt.RoundTrip(req)
net/http.Transport 实现了 net/http.RoundTripper 接口，也是整个请求过程中最重要并且最复杂的结构体，该结构体会在 net/http.Transport.roundTrip 中发送 HTTP 请求并等待响应，我们可以将该函数的执行过程分成两个部分：
- 根据 URL 的协议查找并执行自定义的
net/http.RoundTripper实现； - 从连接池中获取或者初始化新的持久连接并调用连接的
net/http.persistConn.roundTrip发出请求；
// roundTrip implements a RoundTripper over HTTP.
func (t *Transport) roundTrip(req *Request) (*Response, error) {
t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)//初始化TLSNextProto http2使用
ctx := req.Context()//获取请求的上下文
trace := httptrace.ContextClientTrace(ctx)
if req.URL == nil {//错误处理
req.closeBody()
return nil, errors.New("http: nil Request.URL")
}
if req.Header == nil {
req.closeBody()
return nil, errors.New("http: nil Request.Header")
}
scheme := req.URL.Scheme
isHTTP := scheme == "http" || scheme == "https"
if isHTTP {//如果是http或https请求，对Header中的数据进行校验
for k, vv := range req.Header {
if !httpguts.ValidHeaderFieldName(k) {
req.closeBody()
return nil, fmt.Errorf("net/http: invalid header field name %q", k)
}
for _, v := range vv {
if !httpguts.ValidHeaderFieldValue(v) {
req.closeBody()
// Don't include the value in the error, because it may be sensitive.
return nil, fmt.Errorf("net/http: invalid header field value for %q", k)
}
}
}
}
origReq := req
cancelKey := cancelKey{origReq}
req = setupRewindBody(req)
//如果该scheme有自定义的RoundTrip，则使用自定义的RoundTrip处理request，并返回response
if altRT := t.alternateRoundTripper(req); altRT != nil {
if resp, err := altRT.RoundTrip(req); err != ErrSkipAltProtocol {
return resp, err
}
var err error
req, err = rewindBody(req)
if err != nil {
return nil, err
}
}
if !isHTTP {//如果不是http请求，则关闭并退出
req.closeBody()
return nil, badStringError("unsupported protocol scheme", scheme)
}
if req.Method != "" && !validMethod(req.Method) { //对请求的Method进行校验
req.closeBody()
return nil, fmt.Errorf("net/http: invalid method %q", req.Method)
}
if req.URL.Host == "" {//请求的host为空，则返回
req.closeBody()
return nil, errors.New("http: no Host in request URL")
}
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.
//初始化transportRequest,transportRequest是request的包装器
treq := &transportRequest{Request: req, trace: trace, cancelKey: cancelKey}
//根据用户的请求信息获取connectMethod cm
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.
//从缓存中获取一个连接，或者新建一个连接
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 {
resp, err = pconn.roundTrip(treq)
}
if err == nil {
resp.Request = origReq
return resp, nil
}
// Failed. Clean up and determine whether to retry.
if http2isNoCachedConnError(err) {
if t.removeIdleConn(pconn) {
t.decConnsPerHost(pconn.cacheKey)
}
} else if !pconn.shouldRetryRequest(req, err) {
// Issue 16465: return underlying net.Conn.Read error from peek,
// as we've historically done.
if e, ok := err.(nothingWrittenError); ok {
err = e.error
}
if e, ok := err.(transportReadFromServerError); ok {
err = e.err
}
if b, ok := req.Body.(*readTrackingBody); ok && !b.didClose {
// Issue 49621: Close the request body if pconn.roundTrip
// didn't do so already. This can happen if the pconn
// write loop exits without reading the write request.
req.closeBody()
}
return nil, err
}
testHookRoundTripRetried()
// Rewind the body if we're able to.
req, err = rewindBody(req)
if err != nil {
return nil, err
}
}
}getConn用于返回一条长连接。长连接的来源有2种路径：连接池中获取；当连接池中无法获取到时会新建一条连接
func (t *Transport) getConn(treq *transportRequest, cm connectMethod) (pc *persistConn, err error) {
req := treq.Request
trace := treq.trace
ctx := req.Context()
if trace != nil && trace.GetConn != nil {
trace.GetConn(cm.addr())
}
w := &wantConn{
cm: cm,
key: cm.key(),
ctx: ctx,
ready: make(chan struct{}, 1),
beforeDial: testHookPrePendingDial,
afterDial: testHookPostPendingDial,
}
defer func() {
if err != nil {
w.cancel(t, err)
}
}()
// 从连接池中找一条合适的连接，如果找到则返回该连接，否则新建连接
if delivered := t.queueForIdleConn(w); delivered {
pc := w.pc
// Trace only for HTTP/1.
// HTTP/2 calls trace.GotConn itself.
if pc.alt == nil && trace != nil && trace.GotConn != nil {
trace.GotConn(pc.gotIdleConnTrace(pc.idleAt))
}
// set request canceler to some non-nil function so we
// can detect whether it was cleared between now and when
// we enter roundTrip
t.setReqCanceler(treq.cancelKey, func(error) {})
return pc, nil
}
cancelc := make(chan error, 1)
t.setReqCanceler(treq.cancelKey, func(err error) { cancelc <- err })
// 排队等待获取连接
t.queueForDial(w)
// 通过select监听获取连接完成或者取消
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()})
}
if w.err != nil {
// If the request has been cancelled, that's probably
// what caused w.err; if so, prefer to return the
// cancellation error (see golang.org/issue/16049).
select {
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
default:
// return below
}
}
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
}
}排队等待新建连接
// dialConnFor dials on behalf of w and delivers the result to w.
// dialConnFor has received permission to dial w.cm and is counted in t.connCount[w.cm.key()].
// If the dial is canceled or unsuccessful, dialConnFor decrements t.connCount[w.cm.key()].
func (t *Transport) dialConnFor(w *wantConn) {
defer w.afterDial()
pc, err := t.dialConn(w.ctx, w.cm)
delivered := w.tryDeliver(pc, err)
if err == nil && (!delivered || pc.alt != nil) {
// pconn was not passed to w,
// or it is HTTP/2 and can be shared.
// Add to the idle connection pool.
t.putOrCloseIdleConn(pc)
}
if err != nil {
t.decConnsPerHost(w.key)
}
}
// queueForDial queues w to wait for permission to begin dialing.
// Once w receives permission to dial, it will do so in a separate goroutine.
func (t *Transport) queueForDial(w *wantConn) {
w.beforeDial()
if t.MaxConnsPerHost <= 0 {
go t.dialConnFor(w)
return
}
t.connsPerHostMu.Lock()
defer t.connsPerHostMu.Unlock()
if n := t.connsPerHost[w.key]; n < t.MaxConnsPerHost {
if t.connsPerHost == nil {
t.connsPerHost = make(map[connectMethodKey]int)
}
t.connsPerHost[w.key] = n + 1
go t.dialConnFor(w)
return
}
if t.connsPerHostWait == nil {
t.connsPerHostWait = make(map[connectMethodKey]wantConnQueue)
}
q := t.connsPerHostWait[w.key]
q.cleanFront()
q.pushBack(w)
t.connsPerHostWait[w.key] = q
}dialConnFor函数，执行新建连接，拨号功能，如果新建连接成功，则添加当前连接到连接池
// dialConnFor dials on behalf of w and delivers the result to w.
// dialConnFor has received permission to dial w.cm and is counted in t.connCount[w.cm.key()].
// If the dial is canceled or unsuccessful, dialConnFor decrements t.connCount[w.cm.key()].
func (t *Transport) dialConnFor(w *wantConn) {
defer w.afterDial()
// 执行新建连接，拨号功能，如果新建连接成功，则添加当前连接到连接池
pc, err := t.dialConn(w.ctx, w.cm)
delivered := w.tryDeliver(pc, err)
if err == nil && (!delivered || pc.alt != nil) {
// pconn was not passed to w,
// or it is HTTP/2 and can be shared.
// Add to the idle connection pool.
t.putOrCloseIdleConn(pc)
}
if err != nil { // 如果建立连接或者获取连接失败，则删除连接池中的连接。
t.decConnsPerHost(w.key)
}
}dialConn用于新创建一条连接，并为该连接启动readLoop和writeLoop
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{}),
}
trace := httptrace.ContextClientTrace(ctx)
wrapErr := func(err error) error {
if cm.proxyURL != nil {
// Return a typed error, per Issue 16997
return &net.OpError{Op: "proxyconnect", Net: "tcp", Err: err}
}
return err
}
// 调用注册的DialTLS处理tls。使用自注册的TLS处理函数时，transport的TLSClientConfig和TLSHandshakeTimeout
if cm.scheme() == "https" && t.hasCustomTLSDialer() {
var err error
pconn.conn, err = t.customDialTLS(ctx, "tcp", cm.addr())
if err != nil {
return nil, wrapErr(err)
}
// 如果连接类型是TLS的，则需要处理TLS协商
if tc, ok := pconn.conn.(*tls.Conn); ok {
// Handshake here, in case DialTLS didn't. TLSNextProto below
// depends on it for knowing the connection state.
if trace != nil && trace.TLSHandshakeStart != nil {
trace.TLSHandshakeStart()
}
// 启动TLS协商，如果协商失败需要 关闭连接
if err := tc.Handshake(); err != nil {
go pconn.conn.Close()
if trace != nil && trace.TLSHandshakeDone != nil {
trace.TLSHandshakeDone(tls.ConnectionState{}, err)
}
return nil, err
}
cs := tc.ConnectionState()
if trace != nil && trace.TLSHandshakeDone != nil {
trace.TLSHandshakeDone(cs, nil)
}
pconn.tlsState = &cs
}
} else {
// 使用默认方式创建连接,此时会用到transport的TLSClientConfig和TLSHandshakeTimeout参数。同样注意cm.addr()
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的情况
switch {
case cm.proxyURL == nil:
// Do nothing. Not using a proxy.
case cm.proxyURL.Scheme == "socks5":
conn := pconn.conn
d := socksNewDialer("tcp", conn.RemoteAddr().String())
if u := cm.proxyURL.User; u != nil {
auth := &socksUsernamePassword{
Username: u.Username(),
}
auth.Password, _ = u.Password()
d.AuthMethods = []socksAuthMethod{
socksAuthMethodNotRequired,
socksAuthMethodUsernamePassword,
}
d.Authenticate = auth.Authenticate
}
if _, err := d.DialWithConn(ctx, conn, "tcp", cm.targetAddr); err != nil {
conn.Close()
return nil, err
}
case cm.targetScheme == "http":
pconn.isProxy = true
if pa := cm.proxyAuth(); pa != "" {
pconn.mutateHeaderFunc = func(h Header) {
h.Set("Proxy-Authorization", pa)
}
}
case cm.targetScheme == "https":
conn := pconn.conn
hdr := t.ProxyConnectHeader
if hdr == nil {
hdr = make(Header)
}
if pa := cm.proxyAuth(); pa != "" {
hdr = hdr.Clone()
hdr.Set("Proxy-Authorization", pa)
}
connectReq := &Request{
Method: "CONNECT",
URL: &url.URL{Opaque: cm.targetAddr},
Host: cm.targetAddr,
Header: hdr,
}
// If there's no done channel (no deadline or cancellation
// from the caller possible), at least set some (long)
// timeout here. This will make sure we don't block forever
// and leak a goroutine if the connection stops replying
// after the TCP connect.
connectCtx := ctx
if ctx.Done() == nil {
newCtx, cancel := context.WithTimeout(ctx, 1*time.Minute)
defer cancel()
connectCtx = newCtx
}
didReadResponse := make(chan struct{}) // closed after CONNECT write+read is done or fails
var (
resp *Response
err error // write or read error
)
// Write the CONNECT request & read the response.
go func() {
defer close(didReadResponse)
err = connectReq.Write(conn)
if err != nil {
return
}
// Okay to use and discard buffered reader here, because
// TLS server will not speak until spoken to.
br := bufio.NewReader(conn)
resp, err = ReadResponse(br, connectReq)
}()
select {
case <-connectCtx.Done():
conn.Close()
<-didReadResponse
return nil, connectCtx.Err()
case <-didReadResponse:
// resp or err now set
}
if err != nil {
conn.Close()
return nil, err
}
if resp.StatusCode != 200 {
f := strings.SplitN(resp.Status, " ", 2)
conn.Close()
if len(f) < 2 {
return nil, errors.New("unknown status code")
}
return nil, errors.New(f[1])
}
}
if cm.proxyURL != nil && cm.targetScheme == "https" {
if err := pconn.addTLS(cm.tlsHost(), trace); err != nil {
return nil, err
}
}
if s := pconn.tlsState; s != nil && s.NegotiatedProtocolIsMutual && s.NegotiatedProtocol != "" {
if next, ok := t.TLSNextProto[s.NegotiatedProtocol]; ok {
alt := next(cm.targetAddr, pconn.conn.(*tls.Conn))
if e, ok := alt.(http2erringRoundTripper); ok {
// pconn.conn was closed by next (http2configureTransport.upgradeFn).
return nil, e.err
}
return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: alt}, nil
}
}
pconn.br = bufio.NewReaderSize(pconn, t.readBufferSize())
pconn.bw = bufio.NewWriterSize(persistConnWriter{pconn}, t.writeBufferSize())
// 处理请求response
go pconn.readLoop()
// 开启协程处理请求
go pconn.writeLoop()
return pconn, nil
}readLoop循环接收response响应，成功获得response后会将连接返回连接池，便于后续复用。
func (pc *persistConn) readLoop() {
// 当writeLoop或readLoop(异常)跳出循环后，都需要关闭底层连接。即一条连接包含writeLoop和readLoop两个处理，任何一个loop退出(协议升级除外)则该连接不可用,readLoop跳出循环的正常原因是连接上没有待处理的请求，此时关闭连接，释放资源
closeErr := errReadLoopExiting // default value, if not changed below
defer func() {
pc.close(closeErr)
pc.t.removeIdleConn(pc)
}()
// 尝试将连接放回连接池
tryPutIdleConn := func(trace *httptrace.ClientTrace) bool {
if err := pc.t.tryPutIdleConn(pc); err != nil {
closeErr = err
if trace != nil && trace.PutIdleConn != nil && err != errKeepAlivesDisabled {
trace.PutIdleConn(err)
}
return false
}
if trace != nil && trace.PutIdleConn != nil {
trace.PutIdleConn(nil)
}
return true
}
// 变量主要用于阻塞调用者协程读取EOF的resp.body,直到该连接重新放入连接池中。处理逻辑与上面先尝试放入连接池，然后返回response一样，便于连接快速重用
eofc := make(chan struct{})
defer close(eofc) // unblock reader on errors
// Read this once, before loop starts. (to avoid races in tests)
testHookMu.Lock()
testHookReadLoopBeforeNextRead := testHookReadLoopBeforeNextRead
testHookMu.Unlock()
alive := true
for alive {
// 获取允许的response首部的最大字节数
pc.readLimit = pc.maxHeaderResponseSize()
_, err := pc.br.Peek(1)
pc.mu.Lock()
if pc.numExpectedResponses == 0 {
pc.readLoopPeekFailLocked(err)
pc.mu.Unlock()
return
}
pc.mu.Unlock()
rc := <-pc.reqch
trace := httptrace.ContextClientTrace(rc.req.Context())
var resp *Response
// 如果有response数据，则读取并解析为Response格式
if err == nil {
resp, err = pc.readResponse(rc, trace)
} else {
// 可能的错误如server端关闭，发送EOF
err = transportReadFromServerError{err}
closeErr = err
}
if err != nil {
if pc.readLimit <= 0 {
err = fmt.Errorf("net/http: server response headers exceeded %d bytes; aborted", pc.maxHeaderResponseSize())
}
select {
case rc.ch <- responseAndError{err: err}:
case <-rc.callerGone:
return
}
return
}
pc.readLimit = maxInt64 // effectively no limit for response bodies
pc.mu.Lock()
pc.numExpectedResponses--
pc.mu.Unlock()
bodyWritable := resp.bodyIsWritable()
hasBody := rc.req.Method != "HEAD" && resp.ContentLength != 0
if resp.Close || rc.req.Close || resp.StatusCode <= 199 || bodyWritable {
// Don't do keep-alive on error if either party requested a close
// or we get an unexpected informational (1xx) response.
// StatusCode 100 is already handled above.
alive = false
}
if !hasBody || bodyWritable {
pc.t.setReqCanceler(rc.cancelKey, nil)
// Put the idle conn back into the pool before we send the response
// so if they process it quickly and make another request, they'll
// get this same conn. But we use the unbuffered channel 'rc'
// to guarantee that persistConn.roundTrip got out of its select
// potentially waiting for this persistConn to close.
// but after
alive = alive &&
!pc.sawEOF &&
pc.wroteRequest() &&
tryPutIdleConn(trace)
if bodyWritable {
closeErr = errCallerOwnsConn
}
select {
case rc.ch <- responseAndError{res: resp}:
case <-rc.callerGone:
return
}
// Now that they've read from the unbuffered channel, they're safely
// out of the select that also waits on this goroutine to die, so
// we're allowed to exit now if needed (if alive is false)
testHookReadLoopBeforeNextRead()
continue
}
waitForBodyRead := make(chan bool, 2)
body := &bodyEOFSignal{
body: resp.Body,
earlyCloseFn: func() error {
waitForBodyRead <- false
<-eofc // will be closed by deferred call at the end of the function
return nil
},
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
},
}
// 返回的resp.Body类型变为了bodyEOFSignal，如果调用者在读取resp.Body后没有关闭，会导致readLoop阻塞在下面"case bodyEOF := <-waitForBodyRead:"中
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
}
// 此处与处理不带resp.body的场景相同
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:
// 如果读取完response的数据，则该连接可以被重用，否则直接释放。释放一个未读取完数据的连接会导致数据丢失。注意区分bodyEOF和pc.sawEOF的区别，一个是上层通道(http response.Body)关闭，一个是底层通道(TCP)关闭。
pc.t.setReqCanceler(rc.cancelKey, nil) // before pc might return to idle pool
alive = alive &&
bodyEOF &&
!pc.sawEOF &&
pc.wroteRequest() &&
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()
}
}writeLoop用于发送request请求
func (pc *persistConn) writeLoop() {
defer close(pc.writeLoopDone)
// writeLoop会阻塞等待两个IO case 循环等待并处理roundTrip发来的writeRequest数据，此时需要发送request；如果底层连接关闭，则退出writeLoop
for {
select {
case wr := <-pc.writech:
startBytesWritten := pc.nwrite
// 构造request并发送request请求。waitForContinue用于处理首部含"Expect: 100-continue"的request
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)
}
// 请求失败时，需要关闭request和底层连接
if err == nil {
err = pc.bw.Flush()
}
if err != nil {
wr.req.Request.closeBody()
if pc.nwrite == startBytesWritten {
err = nothingWrittenError{err}
}
}
// 将结果发送给readLoop的pc.wroteRequest()函数处理
pc.writeErrCh <- err
// 将结果返回给roundTrip处理，防止响应超时
wr.ch <- err
// 如果发送request失败，需要关闭连接。writeLoop退出时会关闭pc.conn和pc.closech,同时会导致readLoop退出
if err != nil {
pc.close(err)
return
}
case <-pc.closech:
return
}
}
}一个roundTrip用于处理一个request,通过for select来监听结果。
func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err error) {
testHookEnterRoundTrip()
// 此处与getConn中的"t.setReqCanceler(req, func(error) {})"相对应，用于判断request是否被取消, 返回false表示request被取消，不必继续后续请求，关闭连接并返回错误
if !pc.t.replaceReqCanceler(req.cancelKey, pc.cancelRequest) {
pc.t.putOrCloseIdleConn(pc)
return nil, errRequestCanceled
}
pc.mu.Lock()
// 与readLoop配合使用,表示期望的响应的个数
pc.numExpectedResponses++
headerFn := pc.mutateHeaderFunc
pc.mu.Unlock()
if headerFn != nil {
headerFn(req.extraHeaders())
}
// Ask for a compressed version if the caller didn't set their
// own value for Accept-Encoding. We only attempt to
// uncompress the gzip stream if we were the layer that
// requested it.
requestedGzip := false
// 如果需要在request中设置可接受的解码方法，则在request中添加对应的首部。仅支持gzip方式且仅在调用者没有设置这些首部时设置
if !pc.t.DisableCompression &&
req.Header.Get("Accept-Encoding") == "" &&
req.Header.Get("Range") == "" &&
req.Method != "HEAD" {
// Request gzip only, not deflate. Deflate is ambiguous and
// not as universally supported anyway.
// See: https://zlib.net/zlib_faq.html#faq39
//
// Note that we don't request this for HEAD requests,
// due to a bug in nginx:
// https://trac.nginx.org/nginx/ticket/358
// https://golang.org/issue/5522
//
// We don't request gzip if the request is for a range, since
// auto-decoding a portion of a gzipped document will just fail
// anyway. See https://golang.org/issue/8923
requestedGzip = true
req.extraHeaders().Set("Accept-Encoding", "gzip")
}
var continueCh chan struct{}
if req.ProtoAtLeast(1, 1) && req.Body != nil && req.expectsContinue() {
continueCh = make(chan struct{}, 1)
}
// HTTP1.1默认使用长连接，当transport设置DisableKeepAlives时会导致处理每个request时都会新建一个连接。此处的处理逻辑是：如果transport设置了DisableKeepAlives，而request没有设置
if pc.t.DisableKeepAlives && !req.wantsClose() {
req.extraHeaders().Set("Connection", "close")
}
// 用于在异常场景(如request取消)下通知readLoop，roundTrip是否已经退出，防止ReadLoop发送response阻塞
gone := make(chan struct{})
defer close(gone)
defer func() {
if err != nil {
pc.t.setReqCanceler(req.cancelKey, nil)
}
}()
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
// 给writeLoop封装并发送信息，注意此处的先后顺序。首先给writeLoop发送数据，阻塞等待writeLoop接收，待writeLoop接收后才能发送数据给readLoop,因此发送request总会优先接收response
writeErrCh := make(chan error, 1)
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()
// 该循环主要用于处理获取response超时和request取消时的条件跳转。正常情况下收到reponse, 退出roundtrip函数
for {
testHookWaitResLoop()
select {
// writeLoop返回发送request后的结果
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 <-cancelChan:"和”case <-ctxDoneChan:“为request关闭，request关闭也会导致底层连接关闭，但必须处理非上层协议导致底层连接关闭的情况。
case <-pc.closech:
if debugRoundTrip {
req.logf("closech recv: %T %#v", pc.closed, pc.closed)
}
return nil, pc.mapRoundTripError(req, startBytesWritten, pc.closed)
// 等待获取response超时,关闭连接
case <-respHeaderTimer:
if debugRoundTrip {
req.logf("timeout waiting for response headers.")
}
pc.close(errTimeout)
return nil, errTimeout
// 接收到readLoop返回的response结果
case re := <-resc:
// 极异常情况,直接程序panic
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
// request取消
case <-cancelChan:
pc.t.cancelRequest(req.cancelKey, errRequestCanceled)
// 将关闭之后的chan置为nil,用来防止select一直进入该case(close的chan不会阻塞读，读取的数据为0)
cancelChan = nil
case <-ctxDoneChan:
pc.t.cancelRequest(req.cancelKey, req.Context().Err())
cancelChan = nil
ctxDoneChan = nil
}
}
}
- 如果在http.client 中没有设置transport熟悉，则会使用文章开头说的DefaultTransport，这里设置的默认最大空闲连接数MaxIdleConns，每个host最大空闲连接数MaxIdleConnsPerHost是2，每个host的最大连接数MaxConnsPerHost是0。在大量并发情况下，默认配置会造成很多链接，进而性能急剧下降。如果需要控制合适的连接数，就需要使用自定义的client和transport。配置方式如下：
t := http.DefaultTransport.(*http.Transport).Clone()
t.MaxIdleConns = 100
t.MaxConnsPerHost = 100
t.MaxIdleConnsPerHost = 100
httpClient = &http.Client{
Timeout: 10 * time.Second,
Transport: t,
}- 主goroutine ->requestAndChan -> 读循环goroutine
- 主goroutine ->writeRequest-> 写循环goroutine
- 主goroutine 通过select 监听各个channel上的数据， 比如请求取消， timeout，长连接挂了，写流出错，读流出错
记住这里的一个persistConn 是一个client<-----tcp----->server链接（local ip+local srand port(随机port) ------> server ip + server Port）
type Client struct {
// Transport指定执行独立、单次HTTP请求的机制。
// 如果Transport为nil，则使用DefaultTransport。
Transport RoundTripper
// CheckRedirect指定处理重定向的策略。
// 如果CheckRedirect不为nil，客户端会在执行重定向之前调用本函数字段。
// 参数req和via是将要执行的请求和已经执行的请求（切片，越新的请求越靠后）。
// 如果CheckRedirect返回一个错误，本类型的Get方法不会发送请求req，
// 而是返回之前得到的最后一个回复和该错误。（包装进url.Error类型里）
//
// 如果CheckRedirect为nil，会采用默认策略：连续10此请求后停止。
CheckRedirect func(req *Request, via []*Request) error
// Jar指定cookie管理器。
// 如果Jar为nil，请求中不会发送cookie，回复中的cookie会被忽略。
Jar CookieJar
// Timeout指定本类型的值执行请求的时间限制。
// 该超时限制包括连接时间、重定向和读取回复主体的时间。
// 计时器会在Head、Get、Post或Do方法返回后继续运作并在超时后中断回复主体的读取。
//
// Timeout为零值表示不设置超时。
//
// Client实例的Transport字段必须支持CancelRequest方法，
// 否则Client会在试图用Head、Get、Post或Do方法执行请求时返回错误。
// 本类型的Transport字段默认值（DefaultTransport）支持CancelRequest方法。
Timeout time.Duration
}用于http客户端和服务端的结构体
type Request struct {
// Method指定HTTP方法（GET、POST、PUT等）。对客户端，""代表GET。
Method string
// URL在服务端表示被请求的URI，在客户端表示要访问的URL。
//
// 在服务端，URL字段是解析请求行的URI（保存在RequestURI字段）得到的，
// 对大多数请求来说，除了Path和RawQuery之外的字段都是空字符串。
// （参见RFC 2616, Section 5.1.2）
//
// 在客户端，URL的Host字段指定了要连接的服务器，
// 而Request的Host字段（可选地）指定要发送的HTTP请求的Host头的值。
URL *url.URL
// 接收到的请求的协议版本。本包生产的Request总是使用HTTP/1.1
Proto string // "HTTP/1.0"
ProtoMajor int // 1
ProtoMinor int // 0
// Header字段用来表示HTTP请求的头域。如果头域（多行键值对格式）为：
// accept-encoding: gzip, deflate
// Accept-Language: en-us
// Connection: keep-alive
// 则：
// Header = map[string][]string{
// "Accept-Encoding": {"gzip, deflate"},
// "Accept-Language": {"en-us"},
// "Connection": {"keep-alive"},
// }
// HTTP规定头域的键名（头名）是大小写敏感的，请求的解析器通过规范化头域的键名来实现这点。
// 在客户端的请求，可能会被自动添加或重写Header中的特定的头，参见Request.Write方法。
Header Header
// Body是请求的主体。
//
// 在客户端，如果Body是nil表示该请求没有主体买入GET请求。
// Client的Transport字段会负责调用Body的Close方法。
//
// 在服务端，Body字段总是非nil的；但在没有主体时，读取Body会立刻返回EOF。
// Server会关闭请求的主体，ServeHTTP处理器不需要关闭Body字段。
Body io.ReadCloser
// ContentLength记录相关内容的长度。
// 如果为-1，表示长度未知，如果>=0，表示可以从Body字段读取ContentLength字节数据。
// 在客户端，如果Body非nil而该字段为0，表示不知道Body的长度。
ContentLength int64
// TransferEncoding按从最外到最里的顺序列出传输编码，空切片表示"identity"编码。
// 本字段一般会被忽略。当发送或接受请求时，会自动添加或移除"chunked"传输编码。
TransferEncoding []string
// Close在服务端指定是否在回复请求后关闭连接，在客户端指定是否在发送请求后关闭连接。
Close bool
// 在服务端，Host指定URL会在其上寻找资源的主机。
// 根据RFC 2616，该值可以是Host头的值，或者URL自身提供的主机名。
// Host的格式可以是"host:port"。
//
// 在客户端，请求的Host字段（可选地）用来重写请求的Host头。
// 如过该字段为""，Request.Write方法会使用URL字段的Host。
Host string
// Form是解析好的表单数据，包括URL字段的query参数和POST或PUT的表单数据。
// 本字段只有在调用ParseForm后才有效。在客户端，会忽略请求中的本字段而使用Body替代。
Form url.Values
// PostForm是解析好的POST或PUT的表单数据。
// 本字段只有在调用ParseForm后才有效。在客户端，会忽略请求中的本字段而使用Body替代。
PostForm url.Values
// MultipartForm是解析好的多部件表单，包括上传的文件。
// 本字段只有在调用ParseMultipartForm后才有效。
// 在客户端，会忽略请求中的本字段而使用Body替代。
MultipartForm *multipart.Form
// Trailer指定了会在请求主体之后发送的额外的头域。
//
// 在服务端，Trailer字段必须初始化为只有trailer键，所有键都对应nil值。
// （客户端会声明哪些trailer会发送）
// 在处理器从Body读取时，不能使用本字段。
// 在从Body的读取返回EOF后，Trailer字段会被更新完毕并包含非nil的值。
// （如果客户端发送了这些键值对），此时才可以访问本字段。
//
// 在客户端，Trail必须初始化为一个包含将要发送的键值对的映射。（值可以是nil或其终值）
// ContentLength字段必须是0或-1，以启用"chunked"传输编码发送请求。
// 在开始发送请求后，Trailer可以在读取请求主体期间被修改，
// 一旦请求主体返回EOF，调用者就不可再修改Trailer。
//
// 很少有HTTP客户端、服务端或代理支持HTTP trailer。
Trailer Header
// RemoteAddr允许HTTP服务器和其他软件记录该请求的来源地址，一般用于日志。
// 本字段不是ReadRequest函数填写的，也没有定义格式。
// 本包的HTTP服务器会在调用处理器之前设置RemoteAddr为"IP:port"格式的地址。
// 客户端会忽略请求中的RemoteAddr字段。
RemoteAddr string
// RequestURI是被客户端发送到服务端的请求的请求行中未修改的请求URI
// （参见RFC 2616, Section 5.1）
// 一般应使用URI字段，在客户端设置请求的本字段会导致错误。
RequestURI string
// TLS字段允许HTTP服务器和其他软件记录接收到该请求的TLS连接的信息
// 本字段不是ReadRequest函数填写的。
// 对启用了TLS的连接，本包的HTTP服务器会在调用处理器之前设置TLS字段，否则将设TLS为nil。
// 客户端会忽略请求中的TLS字段。
TLS *tls.ConnectionState
}Request类型代表一个服务端接受到的或者客户端发送出去的HTTP请求。
Request各字段的意义和用途在服务端和客户端是不同的。除了字段本身上方文档，还可参见Request.Write方法和RoundTripper接口的文档。
type Response struct {
Status string // 例如"200 OK"
StatusCode int // 例如200
Proto string // 例如"HTTP/1.0"
ProtoMajor int // 例如1
ProtoMinor int // 例如0
// Header保管头域的键值对。
// 如果回复中有多个头的键相同，Header中保存为该键对应用逗号分隔串联起来的这些头的值
// （参见RFC 2616 Section 4.2）
// 被本结构体中的其他字段复制保管的头（如ContentLength）会从Header中删掉。
//
// Header中的键都是规范化的，参见CanonicalHeaderKey函数
Header Header
// Body代表回复的主体。
// Client类型和Transport类型会保证Body字段总是非nil的，即使回复没有主体或主体长度为0。
// 关闭主体是调用者的责任。
// 如果服务端采用"chunked"传输编码发送的回复，Body字段会自动进行解码。
Body io.ReadCloser
// ContentLength记录相关内容的长度。
// 其值为-1表示长度未知（采用chunked传输编码）
// 除非对应的Request.Method是"HEAD"，其值>=0表示可以从Body读取的字节数
ContentLength int64
// TransferEncoding按从最外到最里的顺序列出传输编码，空切片表示"identity"编码。
TransferEncoding []string
// Close记录头域是否指定应在读取完主体后关闭连接。（即Connection头）
// 该值是给客户端的建议，Response.Write方法的ReadResponse函数都不会关闭连接。
Close bool
// Trailer字段保存和头域相同格式的trailer键值对，和Header字段相同类型
Trailer Header
// Request是用来获取此回复的请求
// Request的Body字段是nil（因为已经被用掉了）
// 这个字段是被Client类型发出请求并获得回复后填充的
Request *Request
// TLS包含接收到该回复的TLS连接的信息。 对未加密的回复，本字段为nil。
// 返回的指针是被（同一TLS连接接收到的）回复共享的，不应被修改。
TLS *tls.ConnectionState
}Response代表一个HTTP请求的回复
标签：Body,nil,err,req,pc,go,net,请求 From： https://www.cnblogs.com/codestack/p/18218899