go语言 http模型reactor示例详解

示例

接下来看一段基于reactor的示例。这里运行通过 go run main.go.

然后curl -i 127.0.0.1:8080.效果如下，也是返回了我们期望的结果

在这里插入图片描述

package main

import (
	"flag"
	"fmt"
	"log"
	"strconv"
	"strings"
	"time"
	"unsafe"

	"learn/http/gnet"
)

var res string

type request struct {
	proto, method string
	path, query   string
	head, body    string
	remoteAddr    string
}

type httpServer struct {
	*gnet.EventServer
}

var (
	errMsg      = "Internal Server Error"
	errMsgBytes = []byte(errMsg)
)

type httpCodec struct {
	req request
}

func (hc *httpCodec) Encode(c gnet.Conn, buf []byte) (out []byte, err error) {
	if c.Context() == nil {
		return buf, nil
	}
	return appendResp(out, " Error", "", errMsg+"\n"), nil
}

func (hc *httpCodec) Decode(c gnet.Conn) (out []byte, err error) {
	buf := c.Read()
	c.ResetBuffer()

	// process the pipeline
	var leftover []byte
pipeline:
	leftover, err = parseReq(buf, &hc.req)
	// bad thing happened
	if err != nil {
		c.SetContext(err)
		return nil, err
	} else if len(leftover) == len(buf) {
		// request not ready, yet
		return
	}
	out = appendHandle(out, res)

	buf = leftover
	goto pipeline
}

func (hs *httpServer) OnInitComplete(srv gnet.Server) (action gnet.Action) {
	//log.Printf("HTTP server is listening on %s (multi-cores: %t, loops: %d)\n",
	//	srv.Addr.String(), srv.Multicore, srv.NumEventLoop)
	return
}

func (hs *httpServer) React(frame []byte, c gnet.Conn) (out []byte, action gnet.Action) {
	if c.Context() != nil {
		// bad thing happened
		out = errMsgBytes
		action = gnet.Close
		return
	}
	// handle the request
	out = frame
	return
}

func main() {
	var port int
	var multicore bool

	// Example command: go run http.go --port --multicore=true
	flag.IntVar(&port, "port",, "server port")
	flag.BoolVar(&multicore, "multicore", true, "multicore")
	flag.Parse()

	res = "Hello World!\r\n"

	http := new(httpServer)
	hc := new(httpCodec)

	// Start serving!
	log.Fatal(gnet.Serve(http, fmt.Sprintf("tcp://:%d", port), gnet.WithMulticore(multicore), gnet.WithCodec(hc), gnet.WithNumEventLoop(), gnet.WithReusePort(true)))
}

// appendHandle handles the incoming request and appends the response to
// the provided bytes, which is then returned to the caller.
func appendHandle(b []byte, res string) []byte {
	return appendResp(b, " OK", "", res)
}

// appendResp will append a valid http response to the provide bytes.
// The status param should be the code plus text such as " OK".
// The head parameter should be a series of lines ending with "\r\n" or empty.
func appendResp(b []byte, status, head, body string) []byte {
	b = append(b, "HTTP/.1"...)
	b = append(b, ' ')
	b = append(b, status...)
	b = append(b, '\r', '\n')
	b = append(b, "Server: gnet\r\n"...)
	b = append(b, "Date: "...)
	b = time.Now().AppendFormat(b, "Mon, Jan 2006 15:04:05 GMT")
	b = append(b, '\r', '\n')
	if len(body) > {
		b = append(b, "Content-Length: "...)
		b = strconv.AppendInt(b, int(len(body)), 10)
		b = append(b, '\r', '\n')
	}
	b = append(b, head...)
	b = append(b, '\r', '\n')
	if len(body) > {
		b = append(b, body...)
	}
	return b
}

func bs(b []byte) string {
	return *(*string)(unsafe.Pointer(&b))
}

func parseReq(data []byte, req *request) (leftover []byte, err error) {
	sdata := bs(data)
	var i, s int
	var head string
	var clen int
	q := -
	// method, path, proto line
	for ; i < len(sdata); i++ {
		if sdata[i] == ' ' {
			req.method = sdata[s:i]
			for i, s = i+, i+1; i < len(sdata); i++ {
				if sdata[i] == '?' && q == - {
					q = i - s
				} else if sdata[i] == ' ' {
					if q != - {
						req.path = sdata[s:q]
						req.query = req.path[q+ : i]
					} else {
						req.path = sdata[s:i]
					}
					for i, s = i+, i+1; i < len(sdata); i++ {
						if sdata[i] == '\n' && sdata[i-] == '\r' {
							req.proto = sdata[s:i]
							i, s = i+, i+1
							break
						}
					}
					break
				}
			}
			break
		}
	}
	if req.proto == "" {
		return data, fmt.Errorf("malformed request")
	}
	head = sdata[:s]
	for ; i < len(sdata); i++ {
		if i > && sdata[i] == '\n' && sdata[i-1] == '\r' {
			line := sdata[s : i-]
			s = i +
			if line == "" {
				req.head = sdata[len(head)+ : i+1]
				i++
				if clen > {
					if len(sdata[i:]) < clen {
						break
					}
					req.body = sdata[i : i+clen]
					i += clen
				}
				return data[i:], nil
			}
			if strings.HasPrefix(line, "Content-Length:") {
				n, err := strconv.ParseInt(strings.TrimSpace(line[len("Content-Length:"):]),, 64)
				if err == nil {
					clen = int(n)
				}
			}
		}
	}
	// not enough data
	return data, nil
}

看一下这个源码解析，还是先从gnet.Serve看起来

gnet.Serve

// Serve starts handling events for the specified address.
//
// Address should use a scheme prefix and be formatted
// like `tcp://.168.0.10:9851` or `unix://socket`.
// Valid network schemes:
//  tcp   - bind to both IPv and IPv6
//  tcp  - IPv4
//  tcp  - IPv6
//  udp   - bind to both IPv and IPv6
//  udp  - IPv4
//  udp  - IPv6
//  unix  - Unix Domain Socket
//
// The "tcp" network scheme is assumed when one is not specified.
func Serve(eventHandler EventHandler, protoAddr string, opts ...Option) (err error) {
    // 加载用户指定的配置
	options := loadOptions(opts...)

	logging.Debugf("default logging level is %s", logging.LogLevel())

	var (
		logger logging.Logger
		flush  func() error
	)
	if options.LogPath != "" {
		if logger, flush, err = logging.CreateLoggerAsLocalFile(options.LogPath, options.LogLevel); err != nil {
			return
		}
	} else {
		logger = logging.GetDefaultLogger()
	}
	if options.Logger == nil {
		options.Logger = logger
	}
	defer func() {
		if flush != nil {
			_ = flush()
		}
		logging.Cleanup()
	}()

	// The maximum number of operating system threads that the Go program can use is initially set to,
	// which should also be the maximum amount of I/O event-loops locked to OS threads that users can start up.
	// 为了防止线程过多
	if options.LockOSThread && options.NumEventLoop > {
		logging.Errorf("too many event-loops under LockOSThread mode, should be less than,000 "+
			"while you are trying to set up %d\n", options.NumEventLoop)
		return errors.ErrTooManyEventLoopThreads
	}

	if rbc := options.ReadBufferCap; rbc <= {
		options.ReadBufferCap =x10000
	} else {
		options.ReadBufferCap = internal.CeilToPowerOfTwo(rbc)
	}
	// 解析addr
	network, addr := parseProtoAddr(protoAddr)

    // 初始化listener
	var ln *listener
	if ln, err = initListener(network, addr, options); err != nil {
		return
	}
	defer ln.close()

	return serve(eventHandler, ln, options, protoAddr)
}

可以看出来参数是EventHandler 这样的interface

type (
	// EventHandler represents the server events' callbacks for the Serve call.
	// Each event has an Action return value that is used manage the state
	// of the connection and server.
	EventHandler interface {
		// OnInitComplete fires when the server is ready for accepting connections.
		// The parameter:server has information and various utilities.
		OnInitComplete(server Server) (action Action)

		// OnShutdown fires when the server is being shut down, it is called right after
		// all event-loops and connections are closed.
		OnShutdown(server Server)

		// OnOpened fires when a new connection has been opened.
		// The parameter:c has information about the connection such as it's local and remote address.
		// Parameter:out is the return value which is going to be sent back to the client.
		// It is generally not recommended to send large amounts of data back to the client in OnOpened.
		//
		// Note that the bytes returned by OnOpened will be sent back to client without being encoded.
		OnOpened(c Conn) (out []byte, action Action)

		// OnClosed fires when a connection has been closed.
		// The parameter:err is the last known connection error.
		OnClosed(c Conn, err error) (action Action)

		// PreWrite fires just before any data is written to any client socket, this event function is usually used to
		// put some code of logging/counting/reporting or any prepositive operations before writing data to client.
		PreWrite()

		// React fires when a connection sends the server data.
		// Call c.Read() or c.ReadN(n) within the parameter:c to read incoming data from client.
		// Parameter:out is the return value which is going to be sent back to the client.
		React(frame []byte, c Conn) (out []byte, action Action)

		// Tick fires immediately after the server starts and will fire again
		// following the duration specified by the delay return value.
		Tick() (delay time.Duration, action Action)
	}

	// EventServer is a built-in implementation of EventHandler which sets up each method with a default implementation,
	// you can compose it with your own implementation of EventHandler when you don't want to implement all methods
	// in EventHandler.
	EventServer struct{}
)

initListener

然后看一下初始化监听

func initListener(network, addr string, options *Options) (l *listener, err error) {
	var sockopts []socket.Option
	// 判断是否开启重复使用端口
	if options.ReusePort || strings.HasPrefix(network, "udp") {
		sockopt := socket.Option{SetSockopt: socket.SetReuseport, Opt:}
		sockopts = append(sockopts, sockopt)
	}
 
    // 是否开启nagle算法 默认是关闭
	if options.TCPNoDelay == TCPNoDelay && strings.HasPrefix(network, "tcp") {
		sockopt := socket.Option{SetSockopt: socket.SetNoDelay, Opt:}
		sockopts = append(sockopts, sockopt)
	}
  
    // 设置socket的recv buffer
	if options.SocketRecvBuffer > {
		sockopt := socket.Option{SetSockopt: socket.SetRecvBuffer, Opt: options.SocketRecvBuffer}
		sockopts = append(sockopts, sockopt)
	}
   
    // 设置socket的send buffer
	if options.SocketSendBuffer > {
		sockopt := socket.Option{SetSockopt: socket.SetSendBuffer, Opt: options.SocketSendBuffer}
		sockopts = append(sockopts, sockopt)
	}
	l = &listener{network: network, addr: addr, sockopts: sockopts}
	
	err = l.normalize()
	return
}

normalize最后调用的是tcpSocket方法。

// tcpSocket creates an endpoint for communication and returns a file descriptor that refers to that endpoint.
// Argument `reusePort` indicates whether the SO_REUSEPORT flag will be assigned.
func tcpSocket(proto, addr string, sockopts ...Option) (fd int, netAddr net.Addr, err error) {
	var (
		family   int
		ipvonly bool
		sockaddr unix.Sockaddr
	)
    // 获取地址
	if sockaddr, family, netAddr, ipvonly, err = getTCPSockaddr(proto, addr); err != nil {
		return
	}

    // 调用 底层的socket方法
    // 调用 unix.Socket(family, sotype|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, proto) 
	if fd, err = sysSocket(family, unix.SOCK_STREAM, unix.IPPROTO_TCP); err != nil {
		err = os.NewSyscallError("socket", err)
		return
	}
	defer func() {
		if err != nil {
			_ = unix.Close(fd)
		}
	}()

	if family == unix.AF_INET && ipv6only {
		if err = SetIPvOnly(fd, 1); err != nil {
			return
		}
	}
  
    // 添加率socket的一些自定义参数
	for _, sockopt := range sockopts {
		if err = sockopt.SetSockopt(fd, sockopt.Opt); err != nil {
			return
		}
	}
  
    // bind
	if err = os.NewSyscallError("bind", unix.Bind(fd, sockaddr)); err != nil {
		return
	}
 
    // 设置半连接数量的最大值
	// Set backlog size to the maximum.
	err = os.NewSyscallError("listen", unix.Listen(fd, listenerBacklogMaxSize))

	return
}

serve

func serve(eventHandler EventHandler, listener *listener, options *Options, protoAddr string) error {
	// Figure out the proper number of event-loops/goroutines to run.
	numEventLoop :=
	if options.Multicore {
		numEventLoop = runtime.NumCPU()
	}
	if options.NumEventLoop > {
		numEventLoop = options.NumEventLoop
	}
    // 实例化server
	svr := new(server)
	svr.opts = options
	svr.eventHandler = eventHandler
	svr.ln = listener

    // 判断选择的轮训方式 默认是RoundRobin
	switch options.LB {
	case RoundRobin:
		svr.lb = new(roundRobinLoadBalancer)
	case LeastConnections:
		svr.lb = new(leastConnectionsLoadBalancer)
	case SourceAddrHash:
		svr.lb = new(sourceAddrHashLoadBalancer)
	}

	svr.cond = sync.NewCond(&sync.Mutex{})
	if svr.opts.Ticker {
		svr.tickerCtx, svr.cancelTicker = context.WithCancel(context.Background())
	}
	svr.codec = func() ICodec {
		if options.Codec == nil {
			return new(BuiltInFrameCodec)
		}
		return options.Codec
	}()

	server := Server{
		svr:          svr,
		Multicore:    options.Multicore,
		Addr:         listener.lnaddr,
		NumEventLoop: numEventLoop,
		ReusePort:    options.ReusePort,
		TCPKeepAlive: options.TCPKeepAlive,
	}
	switch svr.eventHandler.OnInitComplete(server) {
	case None:
	case Shutdown:
		return nil
	}
	// 开启svr的start
	if err := svr.start(numEventLoop); err != nil {
		svr.closeEventLoops()
		svr.opts.Logger.Errorf("gnet server is stopping with error: %v", err)
		return err
	}
	defer svr.stop(server)

	allServers.Store(protoAddr, svr)

	return nil
}

func (svr *server) start(numEventLoop int) error {
	if svr.opts.ReusePort || svr.ln.network == "udp" {
	    // 启动eventLoops的事件循环
		return svr.activateEventLoops(numEventLoop)
	}

	return svr.activateReactors(numEventLoop)
}

然后看一下activateEventLoops方法。

activateEventLoops

func (svr *server) activateEventLoops(numEventLoop int) (err error) {
	var striker *eventloop
	// Create loops locally and bind the listeners.
	for i :=; i < numEventLoop; i++ {
		ln := svr.ln
		if i > && (svr.opts.ReusePort || ln.network == "udp") {
		    // 再次调用initListener这个方法 生成新的socket
			if ln, err = initListener(svr.ln.network, svr.ln.addr, svr.opts); err != nil {
				return
			}
		}

		var p *netpoll.Poller
		if p, err = netpoll.OpenPoller(); err == nil {
		    // 实例化eventloop
			el := new(eventloop)
			el.ln = ln
			el.svr = svr
			el.poller = p
			el.buffer = make([]byte, svr.opts.ReadBufferCap)
			el.connections = make(map[int]*conn)
			el.eventHandler = svr.eventHandler
			// 添加监听的套接字
			// 注意这里的loopAccept是一个回调函数
			_ = el.poller.AddRead(el.ln.packPollAttachment(el.loopAccept))
			// 注册
			svr.lb.register(el)

			// Start the ticker.
			if el.idx == && svr.opts.Ticker {
				striker = el
			}
		} else {
			return
		}
	}

	// Start event-loops in background.
	svr.startEventLoops()

	go striker.loopTicker(svr.tickerCtx)

	return
}

然后看一下 OpenPoller方法

// OpenPoller instantiates a poller.
func OpenPoller() (poller *Poller, err error) {
    // 创建poller实例
	poller = new(Poller)
  
    // 调用 epoll_create
	if poller.fd, err = unix.EpollCreate(unix.EPOLL_CLOEXEC); err != nil {
		poller = nil
		err = os.NewSyscallError("epoll_create", err)
		return
	}
 
    // 创建eventfd用来唤醒epoll
	if poller.wfd, err = unix.Eventfd(, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC); err != nil {
		_ = poller.Close()
		poller = nil
		err = os.NewSyscallError("eventfd", err)
		return
	}
	poller.wfdBuf = make([]byte,)
 
    // eventfd加入到监听中
	if err = poller.AddRead(&PollAttachment{FD: poller.wfd}); err != nil {
		_ = poller.Close()
		poller = nil
		return
	}
	// 实例化asyncTaskQueue和priorAsyncTaskQueue
	poller.asyncTaskQueue = queue.NewLockFreeQueue()
	poller.priorAsyncTaskQueue = queue.NewLockFreeQueue()
	return
}

然后看一下loopAccept 这个方法

func (el *eventloop) loopAccept(_ netpoll.IOEvent) error {
	if el.ln.network == "udp" {
		return el.loopReadUDP(el.ln.fd)
	}
    // 因为前面在initListener这里只运行了bind方法 所以这里accept
	nfd, sa, err := unix.Accept(el.ln.fd)
	if err != nil {
		if err == unix.EAGAIN {
			return nil
		}
		el.getLogger().Errorf("Accept() fails due to error: %v", err)
		return os.NewSyscallError("accept", err)
	}
	// 获取到了以后设置为非阻塞
	if err = os.NewSyscallError("fcntl nonblock", unix.SetNonblock(nfd, true)); err != nil {
		return err
	}

	netAddr := socket.SockaddrToTCPOrUnixAddr(sa)
	if el.svr.opts.TCPKeepAlive > && el.svr.ln.network == "tcp" {
		err = socket.SetKeepAlive(nfd, int(el.svr.opts.TCPKeepAlive/time.Second))
		logging.LogErr(err)
	}
  
    // 根据套接字实例化连接
	c := newTCPConn(nfd, el, sa, netAddr)
	// 在epoll中添加监听
	if err = el.poller.AddRead(c.pollAttachment); err == nil {
		el.connections[c.fd] = c
		return el.loopOpen(c)
	}
	return err
}

然后看一下 startEventLoops 这个方法

func (svr *server) startEventLoops() {
    // iterate 就是运行下面的方法
	svr.lb.iterate(func(i int, el *eventloop) bool {
		svr.wg.Add()
		go func() {
		    // 调用loopRun
			el.loopRun(svr.opts.LockOSThread)
			svr.wg.Done()
		}()
		return true
	})
}

func (el *eventloop) loopRun(lockOSThread bool) {
	if lockOSThread {
		runtime.LockOSThread()
		defer runtime.UnlockOSThread()
	}

	defer func() {
		el.closeAllConns()
		el.ln.close()
		el.svr.signalShutdown()
	}()
	// 调用Polling 注意这里Polling里面传的是一个方法
	err := el.poller.Polling(func(fd int, ev uint) (err error) {
	    // 注意里面这个连接有事件发生的时候
		if c, ok := el.connections[fd]; ok {
			// Don't change the ordering of processing EPOLLOUT | EPOLLRDHUP / EPOLLIN unless you're%
			// sure what you're doing!
			// Re-ordering can easily introduce bugs and bad side-effects, as I found out painfully in the past.

			// We should always check for the EPOLLOUT event first, as we must try to send the leftover data back to
			// client when any error occurs on a connection.
			//
			// Either an EPOLLOUT or EPOLLERR event may be fired when a connection is refused.
			// In either case loopWrite() should take care of it properly:
			//) writing data back,
			//) closing the connection.
			if ev&netpoll.OutEvents != && !c.outboundBuffer.IsEmpty() {
			    // 写事件
				if err := el.loopWrite(c); err != nil {
					return err
				}
			}
			// If there is pending data in outbound buffer, then we should omit this readable event
			// and prioritize the writable events to achieve a higher performance.
			//
			// Note that the client may send massive amounts of data to server by write() under blocking mode,
			// resulting in that it won't receive any responses before the server read all data from client,
			// in which case if the socket send buffer is full, we need to let it go and continue reading the data
			// to prevent blocking forever.
			// 读事件
			if ev&netpoll.InEvents != && (ev&netpoll.OutEvents == 0 || c.outboundBuffer.IsEmpty()) {
				return el.loopRead(c)
			}
			return nil
		}
		// 说明只是可以建立新的连接
		return el.loopAccept(ev)
	})
	el.getLogger().Debugf("event-loop(%d) is exiting due to error: %v", el.idx, err)
}

polling

这个方法是比较重要的，也是阻塞在epoll上面,去监听fd的事件

// Polling blocks the current goroutine, waiting for network-events.
func (p *Poller) Polling(callback func(fd int, ev uint) error) error {
	el := newEventList(InitPollEventsCap)
	var wakenUp bool

	msec := -
	for {
	    // 使用epoll_wait
		n, err := unix.EpollWait(p.fd, el.events, msec)
		if n == || (n < 0 && err == unix.EINTR) {
			msec = -
			runtime.Gosched()
			continue
		} else if err != nil {
			logging.Errorf("error occurs in epoll: %v", os.NewSyscallError("epoll_wait", err))
			return err
		}
		msec =
        // 判断每个套接字的事件
		for i :=; i < n; i++ {
			ev := &el.events[i]
            // 判断是不是唤醒的
			if fd := int(ev.Fd); fd != p.wfd {
				switch err = callback(fd, ev.Events); err {
				case nil:
				case errors.ErrAcceptSocket, errors.ErrServerShutdown:
					return err
				default:
					logging.Warnf("error occurs in event-loop: %v", err)
				}
			} else { // poller is awaken to run tasks in queues.
				wakenUp = true
				_, _ = unix.Read(p.wfd, p.wfdBuf)
			}
		}
        // 进行唤醒
		if wakenUp {
			wakenUp = false
			task := p.priorAsyncTaskQueue.Dequeue()
			// 运行任务
			for ; task != nil; task = p.priorAsyncTaskQueue.Dequeue() {
				switch err = task.Run(task.Arg); err {
				case nil:
				case errors.ErrServerShutdown:
					return err
				default:
					logging.Warnf("error occurs in user-defined function, %v", err)
				}
				// 放入任务
				queue.PutTask(task)
			}
			for i :=; i < MaxAsyncTasksAtOneTime; i++ {
				if task = p.asyncTaskQueue.Dequeue(); task == nil {
					break
				}
				switch err = task.Run(task.Arg); err {
				case nil:
				case errors.ErrServerShutdown:
					return err
				default:
					logging.Warnf("error occurs in user-defined function, %v", err)
				}
				queue.PutTask(task)
			}
			atomic.StoreInt(&p.netpollWakeSig, 0)
			if (!p.asyncTaskQueue.Empty() || !p.priorAsyncTaskQueue.Empty()) && atomic.CompareAndSwapInt(&p.netpollWakeSig, 0, 1) {
				for _, err = unix.Write(p.wfd, b); err == unix.EINTR || err == unix.EAGAIN; _, err = unix.Write(p.wfd, b) {
				}
			}
		}

		if n == el.size {
			el.expand()
		} else if n < el.size>> {
			el.shrink()
		}
	}
}

这里主要分析的是在reuse port的情况下，根据你开多少线程那么开多少个open poll，这样的话线程数量就是固定的，就不会出现goroutine暴增的情况，同时因为每次accept连接后，便会设置成了非阻塞的，并且不会阻塞在read和write这样的io事件上，通过这些行为保证了整个流程的高可用

go语言 http模型reactor示例详解

目录

示例

initListener

serve

activateEventLoops

polling