快速掌握Go語言HTTP標準庫的實現(xiàn)方法

第快速掌握Go語言HTTP標準庫的實現(xiàn)方法目錄HTTPclientClient結(jié)構(gòu)體初始化請求NewRequest初始化請求Request準備http發(fā)送請求Transport獲取空閑連接queueForIdleConn建立連接queueForDial等待響應(yīng)httpserver監(jiān)聽處理請求Reference本篇文章來分析一下Go語言HTTP標準庫是如何實現(xiàn)的。

本文使用的go的源碼1.15.7

基于HTTP構(gòu)建的服務(wù)標準模型包括兩個端，客戶端(Client)和服務(wù)端(Server)。HTTP請求從客戶端發(fā)出，服務(wù)端接受到請求后進行處理然后將響應(yīng)返回給客戶端。所以http服務(wù)器的工作就在于如何接受來自客戶端的請求，并向客戶端返回響應(yīng)。

一個典型的HTTP服務(wù)應(yīng)該如圖所示：

HTTPclient

在Go中可以直接通過HTTP包的Get方法來發(fā)起相關(guān)請求數(shù)據(jù)，一個簡單例子：

funcmain(){

resp,err:=http.Get("/getname=luozhiyunage=27")

iferr!=nil{

fmt.Println(err)

return

deferresp.Body.Close()

body,_:=ioutil.ReadAll(resp.Body)

fmt.Println(string(body))

}

我們下面通過這個例子來進行分析。

HTTP的Get方法會調(diào)用到DefaultClient的Get方法，DefaultClient是Client的一個空實例，所以最后會調(diào)用到Client的Get方法：

Client結(jié)構(gòu)體

typeClientstruct{

TransportRoundTripper

CheckRedirectfunc(req*Request,via[]*Request)error

JarCookieJar

Timeouttime.Duration

}

Client結(jié)構(gòu)體總共由四個字段組成：

Transport：表示HTTP事務(wù)，用于處理客戶端的請求連接并等待服務(wù)端的響應(yīng)；

CheckRedirect：用于指定處理重定向的策略；

Jar：用于管理和存儲請求中的cookie；

Timeout：指定客戶端請求的最大超時時間，該超時時間包括連接、任何的重定向以及讀取相應(yīng)的時間；

初始化請求

func(c*Client)Get(urlstring)(resp*Response,errerror){

//根據(jù)方法名、URL和請求體構(gòu)建請求

req,err:=NewRequest("GET",url,nil)

iferr!=nil{

returnnil,err

//執(zhí)行請求

returnc.Do(req)

}

我們要發(fā)起一個請求首先需要根據(jù)請求類型構(gòu)建一個完整的請求頭、請求體、請求參數(shù)。然后才是根據(jù)請求的完整結(jié)構(gòu)來執(zhí)行請求。

NewRequest初始化請求

NewRequest會調(diào)用到NewRequestWithContext函數(shù)上。這個函數(shù)會根據(jù)請求返回一個Request結(jié)構(gòu)體，它里面包含了一個HTTP請求所有信息。

Request

Request結(jié)構(gòu)體有很多字段，我這里列舉幾個大家比較熟悉的字段：

NewRequestWithContext

funcNewRequestWithContext(ctxcontext.Context,method,urlstring,bodyio.Reader)(*Request,error){

//parseurl

u,err:=urlpkg.Parse(url)

iferr!=nil{

returnnil,err

rc,ok:=body.(io.ReadCloser)

if!okbody!=nil{

rc=ioutil.NopCloser(body)

u.Host=removeEmptyPort(u.Host)

req:=Request{

ctx:ctx,

Method:method,

URL:u,

Proto:"HTTP/1.1",

ProtoMajor:1,

ProtoMinor:1,

Header:make(Header),

Body:rc,

Host:u.Host,

returnreq,nil

}

NewRequestWithContext函數(shù)會將請求封裝成一個Request結(jié)構(gòu)體并返回。

準備http發(fā)送請求

如上圖所示，Client調(diào)用Do方法處理發(fā)送請求最后會調(diào)用到send函數(shù)中。

func(c*Client)send(req*Request,deadlinetime.Time)(resp*Response,didTimeoutfunc()bool,errerror){

resp,didTimeout,err=send(req,c.transport(),deadline)

iferr!=nil{

returnnil,didTimeout,err

returnresp,nil,nil

}

Transport

Client的send方法在調(diào)用send函數(shù)進行下一步的處理前會先調(diào)用transport方法獲取DefaultTransport實例，該實例如下：

varDefaultTransportRoundTripper=Transport{

//定義HTTP代理策略

Proxy:ProxyFromEnvironment,

DialContext:(net.Dialer{

Timeout:30*time.Second,

KeepAlive:30*time.Second,

DualStack:true,

}).DialContext,

ForceAttemptHTTP2:true,

//最大空閑連接數(shù)

MaxIdleConns:100,

//空閑連接超時時間

IdleConnTimeout:90*time.Second,

//TLS握手超時時間

TLSHandshakeTimeout:10*time.Second,

ExpectContinueTimeout:1*time.Second,

}

Transport實現(xiàn)RoundTripper接口，該結(jié)構(gòu)體會發(fā)送http請求并等待響應(yīng)。

typeRoundTripperinterface{

RoundTrip(*Request)(*Response,error)

}

從RoundTripper接口我們也可以看出，該接口定義的RoundTrip方法會具體的處理請求，處理完畢之后會響應(yīng)Response。

回到我們上面的Client的send方法中，它會調(diào)用send函數(shù)，這個函數(shù)主要邏輯都交給Transport的RoundTrip方法來執(zhí)行。

RoundTrip會調(diào)用到roundTrip方法中：

func(t*Transport)roundTrip(req*Request)(*Response,error){

t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)

ctx:=req.Context()

trace:=httptrace.ContextClientTrace(ctx)

...

for{

select{

case-ctx.Done():

req.closeBody()

returnnil,ctx.Err()

default:

//封裝請求

treq:=transportRequest{Request:req,trace:trace,cancelKey:cancelKey}

cm,err:=t.connectMethodForRequest(treq)

iferr!=nil{

req.closeBody()

returnnil,err

//獲取連接

pconn,err:=t.getConn(treq,cm)

iferr!=nil{

t.setReqCanceler(cancelKey,nil)

req.closeBody()

returnnil,err

//等待響應(yīng)結(jié)果

varresp*Response

ifpconn.alt!=nil{

//HTTP/2path.

t.setReqCanceler(cancelKey,nil)//notcancelablewithCancelRequest

resp,err=pconn.alt.RoundTrip(req)

}else{

resp,err=pconn.roundTrip(treq)

iferr==nil{

resp.Request=origReq

returnresp,nil

}

roundTrip方法會做兩件事情：

調(diào)用Transport的getConn方法獲取連接；在獲取到連接后，調(diào)用persistConn的roundTrip方法等待請求響應(yīng)結(jié)果；獲取連接getConn

getConn有兩個階段：

調(diào)用queueForIdleConn獲取空閑connection；調(diào)用queueForDial等待創(chuàng)建新的connection；

func(t*Transport)getConn(treq*transportRequest,cmconnectMethod)(pc*persistConn,errerror){

req:=treq.Request

trace:=treq.trace

ctx:=req.Context()

iftrace!=niltrace.GetConn!=nil{

trace.GetConn(cm.addr())

//將請求封裝成wantConn結(jié)構(gòu)體

w:=wantConn{

cm:cm,

key:cm.key(),

ctx:ctx,

ready:make(chanstruct{},1),

beforeDial:testHookPrePendingDial,

afterDial:testHookPostPendingDial,

deferfunc(){

iferr!=nil{

w.cancel(t,err)

//獲取空閑連接

ifdelivered:=t.queueForIdleConn(w);delivered{

pc:=w.pc

t.setReqCanceler(treq.cancelKey,func(error){})

returnpc,nil

//創(chuàng)建連接

t.queueForDial(w)

select{

//獲取到連接后進入該分支

case-w.ready:

returnw.pc,w.err

獲取空閑連接queueForIdleConn

成功獲取到空閑connection：

成功獲取connection分為如下幾步：

根據(jù)當前的請求的地址去空閑connection字典中查看存不存在空閑的connection列表；如果能獲取到空閑的connection列表，那么獲取到列表的最后一個connection；返回；

獲取不到空閑connection：

當獲取不到空閑connection時：

根據(jù)當前的請求的地址去空閑connection字典中查看存不存在空閑的connection列表；不存在該請求的connection列表，那么將該wantConn加入到等待獲取空閑connection字典中；

從上面的圖解應(yīng)該就很能看出這一步會怎么操作了，這里簡要的分析一下代碼，讓大家更清楚里面的邏輯：

func(t*Transport)queueForIdleConn(w*wantConn)(deliveredbool){

ift.DisableKeepAlives{

returnfalse

t.idleMu.Lock()

defert.idleMu.Unlock()

t.closeIdle=false

ifw==nil{

returnfalse

//計算空閑連接超時時間

varoldTimetime.Time

ift.IdleConnTimeout0{

oldTime=time.Now().Add(-t.IdleConnTimeout)

//Lookformostrecently-usedidleconnection.

//找到key相同的connection列表

iflist,ok:=t.idleConn[w.key];ok{

stop:=false

delivered:=false

forlen(list)0!stop{

//找到connection列表最后一個

pconn:=list[len(list)-1]

//檢查這個connection是不是等待太久了

tooOld:=!oldTime.IsZero()pconn.idleAt.Round(0).Before(oldTime)

iftooOld{

gopconn.closeConnIfStillIdle()

//該connection被標記為broken或閑置太久continue

ifpconn.isBroken()||tooOld{

list=list[:len(list)-1]

continue

//嘗試將該connection寫入到w中

delivered=w.tryDeliver(pconn,nil)

ifdelivered{

//操作成功，需要將connection從空閑列表中移除

ifpconn.alt!=nil{

}else{

t.idleLRU.remove(pconn)

list=list[:len(list)-1]

stop=true

iflen(list)0{

t.idleConn[w.key]=list

}else{

//如果該key對應(yīng)的空閑列表不存在，那么將該key從字典中移除

delete(t.idleConn,w.key)

ifstop{

returndelivered

//如果找不到空閑的connection

ift.idleConnWait==nil{

t.idleConnWait=make(map[connectMethodKey]wantConnQueue)

//將該wantConn加入到等待獲取空閑connection字典中

q:=t.idleConnWait[w.key]

q.cleanFront()

q.pushBack(w)

t.idleConnWait[w.key]=q

returnfalse

}

上面的注釋已經(jīng)很清楚了，我這里就不再解釋了。

建立連接queueForDial

在獲取不到空閑連接之后，會嘗試去建立連接，從上面的圖大致可以看到，總共分為以下幾個步驟：

在調(diào)用queueForDial方法的時候會校驗MaxConnsPerHost是否未設(shè)置或已達上限；檢驗不通過則將當前的請求放入到connsPerHostWait等待字典中；如果校驗通過那么會異步的調(diào)用dialConnFor方法創(chuàng)建連接；

dialConnFor方法首先會調(diào)用dialConn方法創(chuàng)建TCP連接，然后啟動兩個異步線程來處理讀寫數(shù)據(jù)，然后調(diào)用tryDeliver將連接綁定到wantConn上面。

下面進行代碼分析：

func(t*Transport)queueForDial(w*wantConn){

w.beforeDial()

//小于零說明無限制，異步建立連接

ift.MaxConnsPerHost=0{

got.dialConnFor(w)

return

t.connsPerHostMu.Lock()

defert.connsPerHostMu.Unlock()

//每個host建立的連接數(shù)沒達到上限，異步建立連接

ifn:=t.connsPerHost[w.key];nt.MaxConnsPerHost{

ift.connsPerHost==nil{

t.connsPerHost=make(map[connectMethodKey]int)

t.connsPerHost[w.key]=n+1

got.dialConnFor(w)

return

//每個host建立的連接數(shù)已達到上限，需要進入等待隊列

ift.connsPerHostWait==nil{

t.connsPerHostWait=make(map[connectMethodKey]wantConnQueue)

q:=t.connsPerHostWait[w.key]

q.cleanFront()

q.pushBack(w)

t.connsPerHostWait[w.key]=q

}

這里主要進行參數(shù)校驗，如果最大連接數(shù)限制為零，亦或是每個host建立的連接數(shù)沒達到上限，那么直接異步建立連接。

dialConnFor

func(t*Transport)dialConnFor(w*wantConn){

deferw.afterDial()

//建立連接

pc,err:=t.dialConn(w.ctx,w.cm)

//連接綁定wantConn

delivered:=w.tryDeliver(pc,err)

//建立連接成功，但是綁定wantConn失敗

//那么將該連接放置到空閑連接字典或調(diào)用等待獲取空閑connection字典中的元素執(zhí)行

iferr==nil(!delivered||pc.alt!=nil){

t.putOrCloseIdleConn(pc)

iferr!=nil{

t.decConnsPerHost(w.key)

}

dialConnFor會調(diào)用dialConn進行TCP連接創(chuàng)建，創(chuàng)建完畢之后調(diào)用tryDeliver方法和wantConn進行綁定。

dialConn

func(t*Transport)dialConn(ctxcontext.Context,cmconnectMethod)(pconn*persistConn,errerror){

//創(chuàng)建連接結(jié)構(gòu)體

pconn=persistConn{

t:t,

cacheKey:cm.key(),

reqch:make(chanrequestAndChan,1),

writech:make(chanwriteRequest,1),

closech:make(chanstruct{}),

writeErrCh:make(chanerror,1),

writeLoopDone:make(chanstruct{}),

ifcm.scheme()=="https"t.hasCustomTLSDialer(){

}else{

//建立tcp連接

conn,err:=t.dial(ctx,"tcp",cm.addr())

iferr!=nil{

returnnil,wrapErr(err)

pconn.conn=conn

ifs:=pconn.tlsState;s!=nils.NegotiatedProtocolIsMutuals.NegotiatedProtocol!=""{

ifnext,ok:=t.TLSNextProto[s.NegotiatedProtocol];ok{

alt:=next(cm.targetAddr,pconn.conn.(*tls.Conn))

ife,ok:=alt.(http2erringRoundTripper);ok{

//pconn.connwasclosedbynext(http2configureTransport.upgradeFn).

returnnil,e.err

returnpersistConn{t:t,cacheKey:pconn.cacheKey,alt:alt},nil

pconn.br=bufio.NewReaderSize(pconn,t.readBufferSize())

pconn.bw=bufio.NewWriterSize(persistConnWriter{pconn},t.writeBufferSize())

//為每個連接異步處理讀寫數(shù)據(jù)

gopconn.readLoop()

gopconn.writeLoop()

returnpconn,nil

}

這里會根據(jù)schema的不同設(shè)置不同的連接配置，我上面顯示的是我們常用的HTTP連接的創(chuàng)建過程。對于HTTP來說會建立tcp連接，然后為連接異步處理讀寫數(shù)據(jù)，最后將創(chuàng)建好的連接返回。

等待響應(yīng)

這一部分的內(nèi)容會稍微復(fù)雜一些，但確實非常的有趣。

在創(chuàng)建連接的時候會初始化兩個channel：writech負責(zé)寫入請求數(shù)據(jù)，reqch負責(zé)讀取響應(yīng)數(shù)據(jù)。我們在上面創(chuàng)建連接的時候，也提到了會為連接創(chuàng)建兩個異步循環(huán)readLoop和writeLoop來負責(zé)處理讀寫數(shù)據(jù)。

在獲取到連接之后，會調(diào)用連接的roundTrip方法，它首先會將請求數(shù)據(jù)寫入到writech管道中，writeLoop接收到數(shù)據(jù)之后就會處理請求。

然后roundTrip會將requestAndChan結(jié)構(gòu)體寫入到reqch管道中，然后roundTrip會循環(huán)等待。readLoop讀取到響應(yīng)數(shù)據(jù)之后就會通過requestAndChan結(jié)構(gòu)體中保存的管道將數(shù)據(jù)封裝成responseAndError結(jié)構(gòu)體回寫，這樣roundTrip就可以接受到響應(yīng)數(shù)據(jù)結(jié)束循環(huán)等待并返回。

roundTrip

func(pc*persistConn)roundTrip(req*transportRequest)(resp*Response,errerror){

writeErrCh:=make(chanerror,1)

//將請求數(shù)據(jù)寫入到writech管道中

pc.writech-writeRequest{req,writeErrCh,continueCh}

//用于接收響應(yīng)的管道

resc:=make(chanresponseAndError)

//將用于接收響應(yīng)的管道封裝成requestAndChan寫入到reqch管道中

pc.reqch-requestAndChan{

req:req.Request,

cancelKey:req.cancelKey,

ch:resc,

for{

testHookWaitResLoop()

select{

//接收到響應(yīng)數(shù)據(jù)

casere:=-resc:

if(re.res==nil)==(re.err==nil){

panic(fmt.Sprintf("internalerror:exactlyoneofresorerrshouldbeset;nil=%v",re.res==nil))

ifdebugRoundTrip{

req.logf("rescrecv:%p,%T/%#v",re.res,re.err,re.err)

ifre.err!=nil{

returnnil,pc.mapRoundTripError(req,startBytesWritten,re.err)

//返回響應(yīng)數(shù)據(jù)

returnre.res,nil

}

這里會封裝好writeRequest作為發(fā)送請求的數(shù)據(jù)，并將用于接收響應(yīng)的管道封裝成requestAndChan寫入到reqch管道中，然后循環(huán)等待接受響應(yīng)。

然后writeLoop會進行請求數(shù)據(jù)writeRequest：

func(pc*persistConn)writeLoop(){

deferclose(pc.writeLoopDone)

for{

select{

casewr:=-pc.writech:

startBytesWritten:=pc.nwrite

//向TCP連接中寫入數(shù)據(jù)，并發(fā)送至目標服務(wù)器

err:=wr.req.Request.write(pc.bw,pc.isProxy,wr.req.extra,pc.waitForContinue(wr.continueCh))

case-pc.closech:

return

}

這里會將從writech管道中獲取到的數(shù)據(jù)寫入到TCP連接中，并發(fā)送至目標服務(wù)器。

readLoop

func(pc*persistConn)readLoop(){

closeErr:=errReadLoopExiting//defaultvalue,ifnotchangedbelow

deferfunc(){

pc.close(closeErr)

pc.t.removeIdleConn(pc)

...

alive:=true

foralive{

pc.readLimit=pc.maxHeaderResponseSize()

//獲取roundTrip發(fā)送的結(jié)構(gòu)體

rc:=-pc.reqch

trace:=httptrace.ContextClientTrace(rc.req.Context())

varresp*Response

iferr==nil{

//讀取數(shù)據(jù)

resp,err=pc.readResponse(rc,trace)

}else{

err=transportReadFromServerError{err}

closeErr=err

...

//將響應(yīng)數(shù)據(jù)寫回到管道中

select{

caserc.ch-responseAndError{res:resp}:

case-rc.callerGone:

return

}

這里是從TCP連接中讀取到對應(yīng)的請求響應(yīng)數(shù)據(jù)，通過roundTrip傳入的管道再回寫，然后roundTrip就會接受到數(shù)據(jù)并獲取的響應(yīng)數(shù)據(jù)返回。

httpserver

我這里繼續(xù)以一個簡單的例子作為開頭：

funcHelloHandler(whttp.ResponseWriter,r*http.Request){

fmt.Fprintf(w,"HelloWorld")

funcmain(){

http.HandleFunc("/",HelloHandler)

http.ListenAndServe(":8000",nil)

}

在實現(xiàn)上面我先用一張圖進行簡要的介紹一下：

其實我們從上面例子的方法名就可以知道一些大致的步驟：

注冊處理器到一個hash表中，可以通過鍵值路由匹配；注冊完之后就是開啟循環(huán)監(jiān)聽，每監(jiān)聽到一個連接就會創(chuàng)建一個Goroutine；在創(chuàng)建好的Goroutine里面會循環(huán)的等待接收請求數(shù)據(jù)，然后根據(jù)請求的地址去處理器路由表中匹配對應(yīng)的處理器，然后將請求交給處理器處理；注冊處理器

處理器的注冊如上面的例子所示，是通過調(diào)用HandleFunc函數(shù)來實現(xiàn)的。

HandleFunc函數(shù)會一直調(diào)用到ServeMux的Handle方法中。

func(mux*ServeMux)Handle(patternstring,handlerHandler){

mux.mu.Lock()

defermux.mu.Unlock()

e:=muxEntry{h:handler,pattern:pattern}

mux.m[pattern]=e

ifpattern[len(pattern)-1]=='/'{

mux.es=appendSorted(mux.es,e)

ifpattern[0]!='/'{

mux.hosts=true

}

Handle會根據(jù)路由作為hash表的鍵來保存muxEntry對象，muxEntry封裝了pattern和handler。如果路由表達式以/結(jié)尾，則將對應(yīng)的muxEntry對象加入到[]muxEntry中。

hash表是用于路由精確匹配，[]muxEntry用于部分匹配。

監(jiān)聽

監(jiān)聽是通過調(diào)用ListenAndServe函數(shù)，里面會調(diào)用server的ListenAndServe方法：

func(srv*Server)ListenAndServe()error{

ifsrv.shuttingDown(){

returnErrServerClosed

addr:=srv.Addr

ifaddr==""{

addr=":http"

//監(jiān)聽端口

ln,err:=net.Listen("tcp",addr)

iferr!=nil{

returnerr

//循環(huán)接收監(jiān)聽到的網(wǎng)絡(luò)請求

returnsrv.Serve(ln)

}

Serve

func(srv*Server)Serve(lnet.Listener)error{

baseCtx:=context.Background()

ctx:=context.WithValue(baseCtx,ServerContextKey,srv)

for{

//接收listener過來的網(wǎng)絡(luò)連接

rw,err:=l.Accept()

...

tempDelay=0

c:=srv.newConn(rw)

c.setState(c.rwc,StateNew)

//創(chuàng)建協(xié)程處理連接

goc.serve(connCtx)

}

Serve這個方法里面會用一個循環(huán)去接收監(jiān)聽到的網(wǎng)絡(luò)連接，然后創(chuàng)建協(xié)程處理連接。所以難免就會有一個問題，如果并發(fā)很高的話，可能會一次性創(chuàng)建太多協(xié)程，導(dǎo)致處理不過來的情況。

處理請求

處理請求是通過為每個連接創(chuàng)建goroutine來處理對應(yīng)的請求：

func(c*conn)serve(ctxcontext.Context){

c.remoteAddr=c.rwc.RemoteAddr().String()

ctx=context.WithValue(ctx,LocalAddrCon