fastHttp服务端处理请求的过程

Github 地址 https://github.com/valyala/fasthttp

fastHttp 服务端的处理请求的过程

工作过程

主要代码

设置监听地址 server.go

func (s *Server) ListenAndServe(addr string) error {
	ln, err := net.Listen("tcp4", addr)
	if err != nil {
		return err
	}
	if tcpln, ok := ln.(*net.TCPListener); ok {
		return s.Serve(tcpKeepaliveListener{
			TCPListener:     tcpln,
			keepalive:       s.TCPKeepalive,
			keepalivePeriod: s.TCPKeepalivePeriod,
		})
	}
	return s.Serve(ln)
}

初始化协程池并启动，当接收到请求后，调 wp.Serve 交给协程池去处理请求 server.go

func (s *Server) Serve(ln net.Listener) error {

 .....

 wp := &workerPool{
   WorkerFunc:      s.serveConn,
   MaxWorkersCount: maxWorkersCount,
   LogAllErrors:    s.LogAllErrors,
   Logger:          s.logger(),
   connState:       s.setState,
 }
 wp.Start()

 for {
   if c, err = acceptConn(s, ln, &lastPerIPErrorTime); err != nil {
     wp.Stop()
     return err
   }
   s.setState(c, StateNew)
   atomic.AddInt32(&s.open, 1)
   if !wp.Serve(c) {
     atomic.AddInt32(&s.open, -1)
     s.writeFastError(c, StatusServiceUnavailable,
       "The connection cannot be served because Server.Concurrency limit exceeded")
     c.Close()
     s.setState(c, StateClosed)
   }
 }
 ......
}

获取请求的句柄 server.go

func acceptConn(s *Server, ln net.Listener, lastPerIPErrorTime *time.Time) (net.Conn, error) {
	for {
		c, err := ln.Accept()
		......
		return c, nil
	}
}

协程池处理请求 workerPool.go

// 从协程池中获取某个协程对应的 句柄channel，然后将 3 中获取到的 请求句柄推入channel
func (wp *workerPool) Serve(c net.Conn) bool {
	ch := wp.getCh()
	if ch == nil {
		return false
	}
	ch.ch <- c
	return true
}

// 这个是协程池工作最重要的部分，获取协程池中协程对应的channel
func (wp *workerPool) getCh() *workerChan {
	var ch *workerChan
	createWorker := false

	wp.lock.Lock()
	ready := wp.ready
	n := len(ready) - 1
  // 获取协程句柄channel失败，如果可以新建协程的话就进行新建，如果不可用新建的话，返回的句柄channel为nil，本次请求被拒绝服务
	if n < 0 {
		if wp.workersCount < wp.MaxWorkersCount {
			createWorker = true
			wp.workersCount++
		}
	} else {
    // 获取协程句柄channel 成功
		ch = ready[n]
		ready[n] = nil
		wp.ready = ready[:n]
	}
	wp.lock.Unlock()

	if ch == nil {
		if !createWorker {
			return nil
		}
    // 新建协程句柄，且为之创建协程
		vch := wp.workerChanPool.Get()
		ch = vch.(*workerChan)
		go func() {
			wp.workerFunc(ch)
			wp.workerChanPool.Put(vch)
		}()
	}
	return ch
}

func (wp *workerPool) workerFunc(ch *workerChan) {
	var c net.Conn
	var err error
	for c = range ch.ch {
    //协程的句柄channel 出现nil的时候 当前协程就退出了
		if c == nil {
			break
		}

    // wp.WorkerFunc是在初始化协程池的时候注册的方法，是  s.serveConn
		if err = wp.WorkerFunc(c); err != nil && err != errHijacked {
		  ......
			}
		}
		......
		if !wp.release(ch) {
			break
		}
	}
	.....
  // 协程的句柄channel 出现nil的时候 当前协程就退出了，退出后将work协程数自减
	wp.workersCount--
	......
}

// 请求处理完之后 将当前协程的句柄channel 放入协程池的ready切片中，待下次获取协程的句柄channel的时候进行复用（复用这个句柄channel就相当于复用了对应的协程）
func (wp *workerPool) release(ch *workerChan) bool {
	ch.lastUseTime = time.Now()
	wp.lock.Lock()
	if wp.mustStop {
		wp.lock.Unlock()
		return false
	}
	wp.ready = append(wp.ready, ch)
	wp.lock.Unlock()
	return true
}

实际处理请求的方法 server.go

func (s *Server) serveConn(c net.Conn) (err error) {
	.....
  //上下文对象复用并初始化上下文
  ctx := s.acquireCtx(c)
	......
		//执行在初始化server的时候自定义的 逻辑
		if continueReadingRequest {
			s.Handler(ctx)
		}
	.....
	// 处理http响应
}

高效的原因
1. fastHttp 协程池详情见 2.2.2 、2.2.4
2. fastHttp 对象复用
  1. 协程channle对象复用详见2.2.4 的 getCh() 函数
  2. 上下文对象复用详见2.2.5

注：
本文为自己对fastHttp的理解，如有疑问，欢迎一起讨论交流
如需转载请注明出处：https://www.cnblogs.com/zhuchenglin/p/14358612.html

lin_zone

fastHttp服务端处理请求的过程