net/http源码分析(二)源码

字数统计: 2.2k字   |   阅读时长: 10分
阅读量

这是Golang源码分析之net/http主题的第二篇，我们已经在上一章节了解了TCP通信的基础框架，现在进入正真的源码分析

以下源码基于go1.14.13 windows/amd64进行分析

一、举个栗子

我们写一个C/S例子

service监听1234端口，设置路由为/hello，处理函数为HelloServer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
//service.go

func main() {
	http.HandleFunc("/hello",HelloServer)
	// 监听TCP地址
	// handler通常设为nil,此时会使用DefaultServeMux(默认路由器)
	http.ListenAndServe(":1234",nil)
}
// 处理函数
func HelloServer(resp http.ResponseWriter, req *http.Request) {
	req.ParseForm()
	fmt.Println(req.Form)
	fmt.Println("path",req.URL.Path)
	fmt.Println("This is xc test")
	fmt.Fprintf(resp, "Hello xc!")
}

client设置每1s发送一个请求，并打印服务端的返回值

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// client.go
func main() {
	for  {
		time.Sleep(time.Second)
		client := &http.Client{}
		req, err := http.NewRequest("GET", "http://127.0.0.1:1234/hello", nil)
		if err!=nil {
			fmt.Println("NewRequest:",err)
			return
		}
		resp,err := client.Do(req)
		if err!=nil {
			fmt.Println("client.Do:",err)
		}
		fmt.Println("resp:",resp)
	}
}

现在启动2个进程(先启服务器，再其客户端)

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
// service.go的日志打印
&{GET /hello HTTP/1.1 1 1 map[Accept-Encoding:[gzip] User-Agent:[Go-http-client/1.1]] {} <nil> 0 [] false 127.0.0.1:1234 map[] map[] <nil> map[] 127.0.0.1:58176 /hello
<nil> <nil> <nil> 0xc000184a40}
ServeHTTP 0x654780
map[]
path /hello
This is xc test


// client.go的日志打印  
resp: &{200 OK 
        200 
        HTTP/1.1 
        1 
        1 
        map[Content-Length:[9] Content-Type:[text/plain; charset=utf-8] Date:[Fri, 02 Apr 2021 03:09:11 GMT]] 
        0xc0000a0040 
        9 
        [] 
        false 
        false 
        map[] 
        0xc000184100 
        <nil>
}

查看网络状态，可以看到客户端和服务端已经建立TCP连接，在双端都是ESTABLISHED状态下进行读写。LISTEN是客户端在监听客户端的接入情况。

二、源码分析

1、服务端源码
1.1、http.HandleFunc 将处理请求的函数注册到路由器map中
1
2
3
4
// 将处理函数handler 注册到默认的路由器DefaultServeMux中，服务器监听到对应的路由，会使用该handler处理
func HandleFunc(pattern string, handler func(ResponseWriter, *Request)) {
	DefaultServeMux.HandleFunc(pattern, handler)
}

DefaultServeMux也即ServeMux，可以看下该路由存储的变量

1
2
3
4
5
6
type ServeMux struct {
	mu    sync.RWMutex  // 锁，由于请求涉及到并发处理，因此这里需要一个锁机制
	m     map[string]muxEntry // 路由规则，一个string对应一个mux实体，这里的string就是注册的路由表达式
	es    []muxEntry // slice of entries sorted from longest to shortest.
	hosts bool       // whether any patterns contain hostnames
}

将pattern路由及对应的处理器以hash表形式存储，在HTTP处理请求的时候会查找对应路由的处理器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
func (mux *ServeMux) Handle(pattern string, handler Handler) {
	// 加锁
	mux.mu.Lock()
	defer mux.mu.Unlock()
	...
	if _, exist := mux.m[pattern]; exist {
		panic("http: multiple registrations for " + pattern)
	}

	if mux.m == nil {
		mux.m = make(map[string]muxEntry)
	}
	// 将pattern与函数匹配，在刚才的例子中，是将"/hello"和HelloServer匹配
	e := muxEntry{h: handler, pattern: pattern}
	mux.m[pattern] = e
	if pattern[len(pattern)-1] == '/' {
		mux.es = appendSorted(mux.es, e)
	}

	if pattern[0] != '/' {
		mux.hosts = true
	}
}
1.2、http.ListenAndServe 用来监听 TCP 连接并处理请求
1
2
3
4
func ListenAndServe(addr string, handler Handler) error {
	server := &Server{Addr: addr, Handler: handler}
	return server.ListenAndServe()
}

ListenAndServe监听对应地址的TCP并处理对应的客户端请求

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
func (srv *Server) ListenAndServe() error {
	if srv.shuttingDown() {
		return ErrServerClosed
	}
	addr := srv.Addr
	if addr == "" {
		addr = ":http"
	}
	// TCP协议搭建服务,此时就是第一章节srver创建socket、bind、listen的流程
	// HTTP的listen
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		return err
	}
	// 监听端口
	return srv.Serve(ln)
}

这里我们看一下net.Listen接口，包含了创建socket、bind绑定socket与地址、listen端口操作。

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
func socket(ctx context.Context, net string, family, sotype, proto int, ipv6only bool, laddr, raddr sockaddr, ctrlFn func(string, string, syscall.RawConn) error) (fd *netFD, err error) {
    // syscall.Socket 系统调用创建socket
	s, err := sysSocket(family, sotype, proto)
	if err != nil {
		return nil, err
	}
	...
    // 创建该网络的描述符，完成netFD结构体的填充
	if fd, err = newFD(s, family, sotype, net); err != nil {
		poll.CloseFunc(s)
		return nil, err
	}

	if laddr != nil && raddr == nil {
		switch sotype {
        // 对于TCP 
		case syscall.SOCK_STREAM, syscall.SOCK_SEQPACKET:
            // syscal.bind() + syscall.Listen
			if err := fd.listenStream(laddr, listenerBacklog(), ctrlFn); err != nil {
				fd.Close()
				return nil, err
			}
			return fd, nil
        // UDP的处理
		case syscall.SOCK_DGRAM:
			if err := fd.listenDatagram(laddr, ctrlFn); err != nil {
				fd.Close()
				return nil, err
			}
			return fd, nil
		}
	}
	if err := fd.dial(ctx, laddr, raddr, ctrlFn); err != nil {
		fd.Close()
		return nil, err
	}
	return fd, nil
}

监听客户端的请求，使用for循环一直等待客户端的请求。

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
// 监听客户端的请求
func (srv *Server) Serve(l net.Listener) error {
	...
	var tempDelay time.Duration // how long to sleep on accept failure

	ctx := context.WithValue(baseCtx, ServerContextKey, srv)
	for {
		// 接收客户端请求  rw客户请求的信息
		// Accept从处于 established 状态的连接队列头部取出一个已经完成的连接
		rw, err := l.Accept()
		if err != nil {
			select {
			case <-srv.getDoneChan():
				return ErrServerClosed
			default:
			}
            // 若有网络错误，则延时等待
			if ne, ok := err.(net.Error); ok && ne.Temporary() {
				if tempDelay == 0 {
					tempDelay = 5 * time.Millisecond
				} else {
					tempDelay *= 2
				}
				if max := 1 * time.Second; tempDelay > max {
					tempDelay = max
				}
				srv.logf("http: Accept error: %v; retrying in %v", err, tempDelay)
				time.Sleep(tempDelay)
				continue
			}
			return err
		}
		...
		c := srv.newConn(rw) // 创建一个新的连接
		c.setState(c.rwc, StateNew) // before Serve can return
		go c.serve(connCtx) // 处理客户端的请求
	}
}

处理客户端的请求

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
// Serve a new connection.
func (c *conn) serve(ctx context.Context) {
	c.remoteAddr = c.rwc.RemoteAddr().String()
	ctx = context.WithValue(ctx, LocalAddrContextKey, c.rwc.LocalAddr())
	...

	// HTTP/1.x from here on.

	ctx, cancelCtx := context.WithCancel(ctx)
	c.cancelCtx = cancelCtx
	defer cancelCtx()

	c.r = &connReader{conn: c}
	c.bufr = newBufioReader(c.r)
	c.bufw = newBufioWriterSize(checkConnErrorWriter{c}, 4<<10)

	for {
        // 循环读用户的请求
		w, err := c.readRequest(ctx)
		if c.r.remain != c.server.initialReadLimitSize() {
			// If we read any bytes off the wire, we're active.
			c.setState(c.rwc, StateActive)
		}
		if err != nil {
			... // 错误处理
		}
        ...

		// 处理请求
		serverHandler{c.server}.ServeHTTP(w, w.req)
		// 维测相关
        ...
	}
}

交给DefaultServeMux路由来处理

1
2
3
4
5
6
7
8
9
10
11
func (sh serverHandler) ServeHTTP(rw ResponseWriter, req *Request) {
	handler := sh.srv.Handler
	if handler == nil {
		handler = DefaultServeMux
	}
	if req.RequestURI == "*" && req.Method == "OPTIONS" {
		handler = globalOptionsHandler{}
	}
    // 这里处理就是示例中注册的HelloServer
	handler.ServeHTTP(rw, req)
}
2、客户端源码

http.NewRequest NewRequest根据方法名、URL 和请求体构建请求。该方法封装成HTTP协议请求的格式，不详细展开

client.Do(req) 处理请求

对于Do方法，我们详细展开，根据层层调用最核心接口即是roundTrip。该接口发送HTTP请求并处理Resp

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 (t *Transport) roundTrip(req *Request) (*Response, error) {
	// 请求检查
	...
	for {
		...
		// 获取连接，通过两种方法获取用于发送请求的连接（1、在队列中闲置的连接，2、创建新连接）
		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)
		}
		...
		// Rewind the body if we're able to.
		req, err = rewindBody(req)
		if err != nil {
			return nil, err
		}
	}
}

连接是一种相对比较昂贵的资源，如果在每次发出 HTTP 请求之前都建立新的连接，可能会消耗比较多的时间，带来较大的额外开销，通过连接池对资源进行分配和复用可以有效地提高 HTTP 请求的整体性能，多数的网络库客户端都会采取类似的策略来复用资源。

所以在获取连接(t.getConn的时候，有两种方法。

第一种：选取在队列中闲置的连接

长链接由一个map[[connectMethodKey]][][]*persistConn组成，connectMethodKey即为一个连接的key包含地址等信息，对应value为persistConn为连接的句柄。选取队列中的连接也就是获取一个句柄。若能获取成功，把该连接从队列中以及idleLRU中删除该连接。

第二种：如果空闲的连接池中没有可用的连接，则会调用queueForDial方法新建连接

使用Dial创建TCP连接，也就是发起socket的listen()。连接创建成功后，会开启两个协程，一个用于处理输入流writeLoop，一个用于处理输出流readLoop。分别从 TCP 连接中读取数据或者向 TCP 连接写入数据，从建立连接的过程我们可以发现，如果我们为每一个 HTTP 请求都创建新的连接并启动 Goroutine 处理读写数据，会占用很多的资源。

发送请求之后，roundTrip接口中pconn.roundTrip处理响应

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
func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err error) {
	...
	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,
	}
	// 循环监听
	for {
		testHookWaitResLoop()
		select {
		...
		case re := <-resc:
			...
			if re.err != nil {
				return nil, pc.mapRoundTripError(req, startBytesWritten, re.err)
			}
			return re.res, nil
			...
		}
	}
}

引用

赏

感谢您的认可

支付宝

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true