Go에서 시작된 HTTP 2.0 요청 프로세스 분석(1부)

(*Transport).roundTrip

(*Transport).roundTrip方法会调用t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)初始化TLSClientConfig以及h2transport, 둘 다 HTTP2.0과 밀접한 관련이 있습니다.

TLSClientConfig: 클라이언트가 지원하는 http 프로토콜을 초기화하고 tls 핸드셰이크 중에 서버에 알립니다.

h2transport: 이 요청이 http2인 경우 h2transport가 연결, 요청 및 응답 처리 논리를 대신합니다.

소스 코드를 살펴보겠습니다.

func (t *Transport) onceSetNextProtoDefaults() {
// ...此处省略代码...
	t2, err := http2configureTransport(t)
if err != nil {
		log.Printf("Error enabling Transport HTTP/2 support: %v", err)
return
	}
	t.h2transport = t2

// ...此处省略代码...
}
func http2configureTransport(t1 *Transport) (*http2Transport, error) {
	connPool := new(http2clientConnPool)
	t2 := &http2Transport{
		ConnPool: http2noDialClientConnPool{connPool},
		t1:       t1,
	}
	connPool.t = t2
if err := http2registerHTTPSProtocol(t1, http2noDialH2RoundTripper{t2}); err != nil {
return nil, err
	}
if t1.TLSClientConfig == nil {
		t1.TLSClientConfig = new(tls.Config)
	}
if !http2strSliceContains(t1.TLSClientConfig.NextProtos, "h2") {
		t1.TLSClientConfig.NextProtos = append([]string{"h2"}, t1.TLSClientConfig.NextProtos...)
	}
if !http2strSliceContains(t1.TLSClientConfig.NextProtos, "http/1.1") {
		t1.TLSClientConfig.NextProtos = append(t1.TLSClientConfig.NextProtos, "http/1.1")
	}
	upgradeFn := func(authority string, c *tls.Conn) RoundTripper {
		addr := http2authorityAddr("https", authority)
if used, err := connPool.addConnIfNeeded(addr, t2, c); err != nil {
go c.Close()
return http2erringRoundTripper{err}
		} else if !used {
// Turns out we don't need this c.
// For example, two goroutines made requests to the same host
// at the same time, both kicking off TCP dials. (since protocol
// was unknown)
go c.Close()
		}
return t2
	}
if m := t1.TLSNextProto; len(m) == 0 {
		t1.TLSNextProto = map[string]func(string, *tls.Conn) RoundTripper{
"h2": upgradeFn,
		}
	} else {
		m["h2"] = upgradeFn
	}
return t2, nil
}

저자는 위의 소스 코드를 다음 단계로 간단히 분해합니다.

1. 新建一个http2clientConnPool并复制给t2，以后http2的请求会优先从该连接池中获取连接。

2. 初始化TLSClientConfig，并将支持的h2和http1.1协议添加到TLSClientConfig.NextProtos中。

3. 定义一个h2的upgradeFn存储到t1.TLSNextProto里。

鉴于前一篇文章对新建连接前的步骤有了较为详细的介绍，所以这里直接看和server建立连接的部分源码，即(*Transport).dialConn方法：

func (t *Transport) dialConn(ctx context.Context, cm connectMethod) (pconn *persistConn, err error) {
// ...此处省略代码...
if cm.scheme() == "https" && t.hasCustomTLSDialer() {
// ...此处省略代码...
	} 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 setup.
// ...此处省略代码...

if s := pconn.tlsState; s != nil && s.NegotiatedProtocolIsMutual && s.NegotiatedProtocol != "" {
if next, ok := t.TLSNextProto[s.NegotiatedProtocol]; ok {
return &persistConn{t: t, cacheKey: pconn.cacheKey, alt: next(cm.targetAddr, pconn.conn.(*tls.Conn))}, nil
		}
	}

// ...此处省略代码...
}

笔者对上述的源码描述如下：

1. 调用t.dial(ctx, "tcp", cm.addr())创建TCP连接。

2. 如果是https的请求， 则对请求建立安全的tls传输通道。

3. 서버와 협상된 경우 TLS의 핸드셰이크 상태를 확인하세요NegotiatedProtocol프로토콜은 비어 있지 않으며 클라이언트의 t.TLSNextProto에는 이 프로토콜이 있으며 alt가 비어 있지 않은 영구 연결을 반환합니다(HTTP1.1은 if를 입력하지 않습니다). 상태) . NegotiatedProtocol协议不为空，且client的t.TLSNextProto有该协议，则返回alt不为空的持久连接（HTTP1.1不会进入if条件里）。

笔者对上述的第三点进行展开。经笔者在本地debug验证，当client和server都支持http2时，s.NegotiatedProtocol的值为h2且s.NegotiatedProtocolIsMutual的值为true。

在上面分析http2configureTransport函数时，我们知道TLSNextProto注册了一个key为h2的函数，所以调用next实际就是调用前面的upgradeFn

저자는 위의 세 번째 점을 확장합니다. 작성자의 로컬 디버그 확인에 따르면 클라이언트와 서버 모두 http2를 지원할 때 s.NegotiatedProtocol의 값은 h2및s.NegotiatedProtocolIsMutual 코드 값>은 <code style="box-sizing: border-box;font-family: SFMono-Regular, Consolas, " liberation mono menlo courier monospace em rgba>true. 🎜

위 분석http2configureTransport 함수, 우리는 TLSNextProto는 키를 h2 함수이므로 다음 실제로, 이전 upgradeFn 함수. 🎜

upgradeFn会调用connPool.addConnIfNeeded向http2的连接池添加一个tls传输通道，并最终返回前面已经创建好的t2即http2Transport。

func (p *http2clientConnPool) addConnIfNeeded(key string, t *http2Transport, c *tls.Conn) (used bool, err error) {
	p.mu.Lock()
// ...此处省略代码...
// 主要用于判断是否有必要像连接池添加新的连接
// 判断连接池中是否已有同host连接，如果有且该链接能够处理新的请求则直接返回
	call, dup := p.addConnCalls[key]
if !dup {
// ...此处省略代码...
		call = &http2addConnCall{
			p:    p,
			done: make(chan struct{}),
		}
		p.addConnCalls[key] = call
go call.run(t, key, c)
	}
	p.mu.Unlock()

	<-call.done
if call.err != nil {
return false, call.err
	}
return !dup, nil
}
func (c *http2addConnCall) run(t *http2Transport, key string, tc *tls.Conn) {
	cc, err := t.NewClientConn(tc)

	p := c.p
	p.mu.Lock()
if err != nil {
		c.err = err
	} else {
		p.addConnLocked(key, cc)
	}
delete(p.addConnCalls, key)
	p.mu.Unlock()
close(c.done)
}

分析上述的源码我们能够得到两点结论：

1. 执行完upgradeFn之后，(*Transport).dialConn返回的持久化连接中alt字段已经不是nil了。

2. t.NewClientConn(tc)新建出来的连接会保存在http2的连接池即http2clientConnPool中，下一小结将对NewClientConn展开分析。

最后我们回到(*Transport).roundTrip方法并分析其中的关键源码：

func (t *Transport) roundTrip(req *Request) (*Response, error) {
	t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)
// ...此处省略代码...
for {
select {
case <-ctx.Done():
			req.closeBody()
return nil, ctx.Err()
default:
		}

// ...此处省略代码...
		pconn, err := t.getConn(treq, cm)
if err != nil {
			t.setReqCanceler(req, nil)
			req.closeBody()
return nil, err
		}

var resp *Response
if pconn.alt != nil {
// HTTP/2 path.
			t.setReqCanceler(req, nil) // not cancelable with CancelRequest
			resp, err = pconn.alt.RoundTrip(req)
		} else {
			resp, err = pconn.roundTrip(treq)
		}
if err == nil {
return resp, nil
		}

// ...此处省略代码...
	}
}

结合前面的分析，pconn.alt在server和client都支持http2协议的情况下是不为nil的。所以，http2的请求会走pconn.alt.RoundTrip(req)分支，也就是说http2的请求流程就被http2Transport接管啦。

(*http2Transport).NewClientConn

(*http2Transport).NewClientConn内部会调用t.newClientConn(c, t.disableKeepAlives())。

因为本节内容较多，所以笔者不再一次性贴出源码，而是按关键步骤分析并分块儿贴出源码。

1、初始化一个http2ClientConn：

cc := &http2ClientConn{
	t:                     t,
	tconn:                 c,
	readerDone:            make(chan struct{}),
	nextStreamID:          1,
	maxFrameSize:          16 << 10,           // spec default
	initialWindowSize:     65535,              // spec default
	maxConcurrentStreams:  1000,               // "infinite", per spec. 1000 seems good enough.
	peerMaxHeaderListSize: 0xffffffffffffffff, // "infinite", per spec. Use 2^64-1 instead.
	streams:               make(map[uint32]*http2clientStream),
	singleUse:             singleUse,
	wantSettingsAck:       true,
	pings:                 make(map[[8]byte]chan struct{}),
}

上面的源码新建了一个默认的http2ClientConn。

initialWindowSize：初始化窗口大小为65535，这个值之后会初始化每一个数据流可发送的数据窗口大小。

maxConcurrentStreams：表示每个连接上允许最多有多少个数据流同时传输数据。

streams：当前连接上的数据流。

singleUse: 控制http2的连接是否允许多个数据流共享，其值由t.disableKeepAlives()控制。

2、创建一个条件锁并且新建Writer&Reader。

cc.cond = sync.NewCond(&cc.mu)
cc.flow.add(int32(http2initialWindowSize))
cc.bw = bufio.NewWriter(http2stickyErrWriter{c, &cc.werr})
cc.br = bufio.NewReader(c)

新建Writer&Reader没什么好说的，需要注意的是cc.flow.add(int32(http2initialWindowSize))。

cc.flow.add将当前连接的可写流控制窗口大小设置为http2initialWindowSize,即65535。

3、新建一个读写数据帧的Framer。

cc.fr = http2NewFramer(cc.bw, cc.br)
cc.fr.ReadMetaHeaders = hpack.NewDecoder(http2initialHeaderTableSize, nil)
cc.fr.MaxHeaderListSize = t.maxHeaderListSize()

4、向server发送开场白，并发送一些初始化数据帧。

initialSettings := []http2Setting{
	{ID: http2SettingEnablePush, Val: 0},
	{ID: http2SettingInitialWindowSize, Val: http2transportDefaultStreamFlow},
}
if max := t.maxHeaderListSize(); max != 0 {
	initialSettings = append(initialSettings, http2Setting{ID: http2SettingMaxHeaderListSize, Val: max})
}

cc.bw.Write(http2clientPreface)
cc.fr.WriteSettings(initialSettings...)
cc.fr.WriteWindowUpdate(0, http2transportDefaultConnFlow)
cc.inflow.add(http2transportDefaultConnFlow + http2initialWindowSize)
cc.bw.Flush()

client向server发送的开场白内容如下:

const (
// client首先想server发送以PRI开头的一串字符串。
    http2ClientPreface = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
)
var (
	http2clientPreface = []byte(http2ClientPreface)
)

发送完开场白后，client向server发送SETTINGS数据帧。

http2SettingEnablePush: 告知server客户端是否开启push功能。

http2SettingInitialWindowSize：告知server客户端可接受的最大数据窗口是http2transportDefaultStreamFlow（4M）。

发送完SETTINGS数据帧后，发送WINDOW_UPDATE数据帧, 因为第一个参数为0即streamID为0，则是告知server此连接可接受的最大数据窗口为http2transportDefaultConnFlow（1G）。

发送完WINDOW_UPDATE数据帧后，将client的可读流控制窗口大小设置为http2transportDefaultConnFlow + http2initialWindowSize。

5、开启读循环并返回

go cc.readLoop()

(*http2Transport).RoundTrip

(*http2Transport).RoundTrip只是一个入口函数，它会调用(*http2Transport). RoundTripOpt方法。

(*http2Transport). RoundTripOpt有两个步骤比较关键：

t.connPool().GetClientConn(req, addr): 在http2的连接池里面获取一个可用连接，其中连接池的类型为http2noDialClientConnPool，参考http2configureTransport函数。

cc.roundTrip(req): 通过获取到的可用连接发送请求并返回响应。

(http2noDialClientConnPool).GetClientConn

根据实际的debug结果(http2noDialClientConnPool).GetClientConn最终会调用(*http2clientConnPool).getClientConn(req *Request, addr string, dialOnMiss bool)。

通过(http2noDialClientConnPool).GetClientConn获取连接时传递给(*http2clientConnPool).getClientConn方法的第三个参数始终为false，该参数为false时代表着即使无法正常获取可用连接，也不在这个环节重新发起拨号流程。

在(*http2clientConnPool).getClientConn中会遍历同地址的连接，并判断连接的状态从而获取一个可以处理请求的连接。

for _, cc := range p.conns[addr] {
if st := cc.idleState(); st.canTakeNewRequest {
if p.shouldTraceGetConn(st) {
			http2traceGetConn(req, addr)
		}
		p.mu.Unlock()
return cc, nil
	}
}

cc.idleState()判断当前连接池中的连接能否处理新的请求:

1、当前连接是否能被多个请求共享，如果仅单个请求使用且已经有一个数据流，则当前连接不能处理新的请求。

if cc.singleUse && cc.nextStreamID > 1 {
return
}

2、以下几点均为true时，才代表当前连接能够处理新的请求：

• 连接状态正常，即未关闭并且不处于正在关闭的状态。

• 当前连接正在处理的数据流小于maxConcurrentStreams。

• 下一个要处理的数据流 + 当前连接处于等待状态的请求*2 < math.MaxInt32。

• 当前连接没有长时间处于空闲状态（主要通过cc.tooIdleLocked()判断）。

st.canTakeNewRequest = cc.goAway == nil && !cc.closed && !cc.closing && maxConcurrentOkay &&
int64(cc.nextStreamID)+2*int64(cc.pendingRequests) < math.MaxInt32 &&
		!cc.tooIdleLocked()

当从链接池成功获取到一个可以处理请求的连接，就可以和server进行数据交互，即(*http2ClientConn).roundTrip流程。

(*http2ClientConn).roundTrip

1、在真正开始处理请求前，还要进行header检查，http2对http1.1的某些header是不支持的，笔者就不对这个逻辑进行分析了，直接上源码：

func http2checkConnHeaders(req *Request) error {
if v := req.Header.Get("Upgrade"); v != "" {
return fmt.Errorf("http2: invalid Upgrade request header: %q", req.Header["Upgrade"])
	}
if vv := req.Header["Transfer-Encoding"]; len(vv) > 0 && (len(vv) > 1 || vv[0] != "" && vv[0] != "chunked") {
return fmt.Errorf("http2: invalid Transfer-Encoding request header: %q", vv)
	}
if vv := req.Header["Connection"]; len(vv) > 0 && (len(vv) > 1 || vv[0] != "" && !strings.EqualFold(vv[0], "close") && !strings.EqualFold(vv[0], "keep-alive")) {
return fmt.Errorf("http2: invalid Connection request header: %q", vv)
	}
return nil
}
func http2commaSeparatedTrailers(req *Request) (string, error) {
	keys := make([]string, 0, len(req.Trailer))
for k := range req.Trailer {
		k = CanonicalHeaderKey(k)
switch k {
case "Transfer-Encoding", "Trailer", "Content-Length":
return "", &http2badStringError{"invalid Trailer key", k}
		}
		keys = append(keys, k)
	}
if len(keys) > 0 {
		sort.Strings(keys)
return strings.Join(keys, ","), nil
	}
return "", nil
}

2、调用(*http2ClientConn).awaitOpenSlotForRequest，一直等到当前连接处理的数据流小于maxConcurrentStreams, 如果此函数返回错误，则本次请求失败。

2.1、double check当前连接可用。

if cc.closed || !cc.canTakeNewRequestLocked() {
if waitingForConn != nil {
close(waitingForConn)
	}
return http2errClientConnUnusable
}

2.2、如果当前连接处理的数据流小于maxConcurrentStreams则直接返回nil。笔者相信大部分逻辑走到这儿就返回了。

if int64(len(cc.streams))+1 <= int64(cc.maxConcurrentStreams) {
if waitingForConn != nil {
close(waitingForConn)
	}
return nil
}

2.3、如果当前连接处理的数据流确实已经达到上限，则开始进入等待流程。

if waitingForConn == nil {
	waitingForConn = make(chan struct{})
go func() {
if err := http2awaitRequestCancel(req, waitingForConn); err != nil {
			cc.mu.Lock()
			waitingForConnErr = err
			cc.cond.Broadcast()
			cc.mu.Unlock()
		}
	}()
}
cc.pendingRequests++
cc.cond.Wait()
cc.pendingRequests--

通过上面的逻辑知道，当前连接处理的数据流达到上限后有两种情况，一是等待请求被取消，二是等待其他请求结束。如果有其他数据流结束并唤醒当前等待的请求，则重复2.1、2.2和2.3的步骤。

3、调用cc.newStream()在连接上创建一个数据流（创建数据流是线程安全的，因为源码中在调用awaitOpenSlotForRequest之前先加锁，直到写入请求的header之后才释放锁）。

func (cc *http2ClientConn) newStream() *http2clientStream {
	cs := &http2clientStream{
		cc:        cc,
		ID:        cc.nextStreamID,
		resc:      make(chan http2resAndError, 1),
		peerReset: make(chan struct{}),
		done:      make(chan struct{}),
	}
	cs.flow.add(int32(cc.initialWindowSize))
	cs.flow.setConnFlow(&cc.flow)
	cs.inflow.add(http2transportDefaultStreamFlow)
	cs.inflow.setConnFlow(&cc.inflow)
	cc.nextStreamID += 2
	cc.streams[cs.ID] = cs
return cs
}

笔者对上述代码简单描述如下：

• 新建一个http2clientStream，数据流ID为cc.nextStreamID，新建数据流后，cc.nextStreamID +=2。

• 数据流通过http2resAndError管道接收请求的响应。

• 初始化当前数据流的可写流控制窗口大小为cc.initialWindowSize，并保存连接的可写流控制指针。

• 初始化当前数据流的可读流控制窗口大小为http2transportDefaultStreamFlow，并保存连接的可读流控制指针。

• 最后将新建的数据流注册到当前连接中。

4、调用cc.t.getBodyWriterState(cs, body)会返回一个http2bodyWriterState结构体。通过该结构体可以知道请求body是否发送成功。

func (t *http2Transport) getBodyWriterState(cs *http2clientStream, body io.Reader) (s http2bodyWriterState) {
	s.cs = cs
if body == nil {
return
	}
	resc := make(chan error, 1)
	s.resc = resc
	s.fn = func() {
		cs.cc.mu.Lock()
		cs.startedWrite = true
		cs.cc.mu.Unlock()
		resc <- cs.writeRequestBody(body, cs.req.Body)
	}
	s.delay = t.expectContinueTimeout()
if s.delay == 0 ||
		!httpguts.HeaderValuesContainsToken(
			cs.req.Header["Expect"],
"100-continue") {
return
	}
// 此处省略代码，因为绝大部分请求都不会设置100-continue的标头
return
}

s.fn: 标记当前数据流开始写入数据，并且将请求body的发送结果写入s.resc管道（本文暂不对writeRequestBody展开分析，下篇文章会对其进行分析）。

5、因为是多个请求共享一个连接，那么向连接写入数据帧时需要加锁，比如加锁写入请求头。

cc.wmu.Lock()
endStream := !hasBody && !hasTrailers
werr := cc.writeHeaders(cs.ID, endStream, int(cc.maxFrameSize), hdrs)
cc.wmu.Unlock()

6、如果有请求body，则开始写入请求body，没有请求body则设置响应header的超时时间（有请求body时，响应header的超时时间需要在请求body写完之后设置）。

if hasBody {
	bodyWriter.scheduleBodyWrite()
} else {
	http2traceWroteRequest(cs.trace, nil)
if d := cc.responseHeaderTimeout(); d != 0 {
		timer := time.NewTimer(d)
defer timer.Stop()
		respHeaderTimer = timer.C
	}
}

scheduleBodyWrite的内容如下：

func (s http2bodyWriterState) scheduleBodyWrite() {
if s.timer == nil {
// We're not doing a delayed write (see
// getBodyWriterState), so just start the writing
// goroutine immediately.
go s.fn()
return
	}
	http2traceWait100Continue(s.cs.trace)
if s.timer.Stop() {
		s.timer.Reset(s.delay)
	}
}

因为笔者的请求header中没有携带100-continue标头，所以在前面的getBodyWriterState函数中初始化的s.timer为nil即调用scheduleBodyWrite会立即开始发送请求body。

7、轮询管道获取响应结果。

在看轮询源码之前，先看一个简单的函数：

handleReadLoopResponse := func(re http2resAndError) (*Response, bool, error) {
	res := re.res
if re.err != nil || res.StatusCode > 299 {
		bodyWriter.cancel()
		cs.abortRequestBodyWrite(http2errStopReqBodyWrite)
	}
if re.err != nil {
		cc.forgetStreamID(cs.ID)
return nil, cs.getStartedWrite(), re.err
	}
	res.Request = req
	res.TLS = cc.tlsState
return res, false, nil
}

该函数主要就是判断读到的响应是否正常，并根据响应的结果构造(*http2ClientConn).roundTrip的返回值。

了解了handleReadLoopResponse之后，下面就看看轮询的逻辑：

for {
select {
case re := <-readLoopResCh:
return handleReadLoopResponse(re)
// 此处省略代码（包含请求取消，请求超时等管道的轮询）
case err := <-bodyWriter.resc:
// Prefer the read loop&#39;s response, if available. Issue 16102.
select {
case re := <-readLoopResCh:
return handleReadLoopResponse(re)
default:
		}
if err != nil {
			cc.forgetStreamID(cs.ID)
return nil, cs.getStartedWrite(), err
		}
		bodyWritten = true
if d := cc.responseHeaderTimeout(); d != 0 {
			timer := time.NewTimer(d)
defer timer.Stop()
			respHeaderTimer = timer.C
		}
	}
}

笔者仅对上面的第二种情况即请求body发送完成进行描述：

• 能否读到响应，如果能够读取响应则直接返回。

• 判断请求body是否发送成功，如果发送失败，直接返回。

• 요청 본문이 성공적으로 전송되면 응답 헤더의 시간 초과를 설정하세요.

요약

이 문서에서는 주로 두 가지 측면을 설명합니다.

1. 클라이언트와 서버 모두 http2 프로토콜을 지원하는지 확인하고, http2 연결을 구축하고, 연결의 읽기 루프를 시작합니다.

2. http2 연결 풀을 통해 http2 연결을 얻고 요청을 보내고 응답을 읽습니다.

위 내용은 Go에서 시작된 HTTP 2.0 요청 프로세스 분석(1부)의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!