-
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响应 }
-
-
-
高效的原因
- fastHttp 协程池 详情见 2.2.2 、2.2.4
- fastHttp 对象复用
- 协程channle对象 复用 详见2.2.4 的 getCh() 函数
- 上下文对象复用 详见2.2.5
注:
本文为自己对fastHttp的理解,如有疑问,欢迎一起讨论交流
如需转载请注明出处:https://www.cnblogs.com/zhuchenglin/p/14358612.html