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

巴特西

fastHttp服务端处理请求的过程

最新文章

热门文章