【TCP服务器】仿http.server源码实现服务与逻辑分离

痛定思痛。 2022-10-21 03:49 48阅读 0赞

# 一、HTTP服务的创建和原理图 #

import "net/http"
    func main(){ 
    	// 逻辑
    	mux := http.NewServeMux()
    	mux.HandleFunc("/hello", func(rw http.ResponseWriter, req *http.Request) { 
    		rw.Write([]byte("welcome"))
    	})
    	// 服务
    	server := http.Server{ 
    		Addr:         "192.168.27.128:80",
    		WriteTimeout: time.Second * 3,
    		Handler:      mux,
    	}
    	log.Fatal(server.ListenAndServe())
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70_pic_center]

根据上图得知。主要对象有三个:

*  server：初始化服务、并监听
 *  conn：处理套接字
 *  handler：逻辑处理

创建三个文件分别为tcp/server.go、tcp/conn.go、main/handler.go

# 二、TCP Server 的实现 #

## 2.1 步陬 ##

1.  创建Server结构体，并有ListenAndServe方法
2.  Serve 方法处理请求

## 2.2 代码 ##

package tcp
    
    type contextKey struct { 
    	name string
    }
    func (k *contextKey) String() string { 
    	return "tcp_proxy context value " + k.name
    }
    
    var (
    	ErrServerClosed = errors.New("tcp: Server closed")
    	ErrAbortHandler = errors.New("tcp: Abort TCPHandler")
    	ServerContextKey = &contextKey{ "tcp-server"}
    	LocalAddrContextKey = &contextKey{ "local-addr"}
    )
    
    // onceCloseListener 封装请求监听器
    type onceCloseListener struct { 
    	net.Listener
    	once     sync.Once
    	closeErr error
    }
    
    // 只关闭一次listener, 后续则直接返回关闭错误信息
    func (oc *onceCloseListener) Close() error { 
    	oc.once.Do(oc.close)
    	return oc.closeErr
    }
    
    // sync.Once只接受方法参数
    func (oc *onceCloseListener) close() { 
    	oc.closeErr = oc.Listener.Close()
    }
    
    // Handler ...
    type Handler interface { 
    	ServeTCP(ctx context.Context, conn net.Conn)
    }
    
    // Server ...
    type Server struct { 
    	Addr    string
    	Handler Handler
    	err     error
    	BaseCtx context.Context
    
    	WriteTimeout     time.Duration
    	ReadTimeout      time.Duration
    	KeepAliveTimeout time.Duration
    
    	mu         sync.Mutex
    	inShutdown int32
    	doneChan   chan struct{ }
    	l          *onceCloseListener
    }
    
    // shuttingDown ...
    func (srv *Server) shuttingDown() bool { 
    	return atomic.LoadInt32(&srv.inShutdown) != 0
    }
    
    // ListenAndServe ...
    func (srv *Server) ListenAndServe() error { 
    	if srv.shuttingDown() { 
    		return ErrServerClosed
    	}
    	if srv.doneChan == nil { 
    		srv.doneChan = make(chan struct{ })
    	}
    	addr := srv.Addr
    	if addr == "" { 
    		return errors.New("invalid address")
    	}
    	ln, err := net.Listen("tcp", addr)
    	if err != nil { 
    		return err
    	}
    	tln := tcpKeepAliveListener{ ln.(*net.TCPListener)}
    	return srv.Serve(tln)
    }
    
    // Close 关闭监听器
    func (srv *Server) Close() error { 
    	atomic.StoreInt32(&srv.inShutdown, 1)
    	close(srv.doneChan)
    	srv.l.Close()
    	return nil
    }
    
    // Serve 处理请求
    func (srv *Server) Serve(l net.Listener) error { 
    	srv.l = &onceCloseListener{ Listener: l}
    	defer srv.l.Close() //执行listener关闭
    	if srv.BaseCtx == nil { 
    		srv.BaseCtx = context.Background()
    	}
    	baseCtx := srv.BaseCtx
    	ctx := context.WithValue(baseCtx, ServerContextKey, srv)
    	for { 
    		rw, e := l.Accept()
    		if e != nil { 
    			select { 
    			case <-srv.getDoneChan():
    				return ErrServerClosed
    			default:
    			}
    			fmt.Printf("accept fail, err: %v\n", e)
    			continue
    		}
    		c := srv.newConn(rw)
    		go c.serve(ctx)
    	}
    }
    
    func (srv *Server) getDoneChan() <-chan struct{ } { 
    	srv.mu.Lock()
    	defer srv.mu.Unlock()
    	if srv.doneChan == nil { 
    		srv.doneChan = make(chan struct{ })
    	}
    	return srv.doneChan
    }
    
    func (srv *Server) newConn(rwc net.Conn) *conn { 
    	c := &conn{ 
    		server: srv,
    		rwc:    rwc,
    	}
    	if d := c.server.ReadTimeout; d != 0 { 
    		c.rwc.SetReadDeadline(time.Now().Add(d))
    	}
    	if d := c.server.WriteTimeout; d != 0 { 
    		c.rwc.SetWriteDeadline(time.Now().Add(d))
    	}
    	if d := c.server.KeepAliveTimeout; d != 0 { 
    		if tcpConn, ok := c.rwc.(*net.TCPConn); ok { 
    			tcpConn.SetKeepAlive(true)
    			tcpConn.SetKeepAlivePeriod(d)
    		}
    	}
    	return c
    }
    
    type tcpKeepAliveListener struct { 
    	*net.TCPListener
    }
    
    // 继承tcpListenner并重写Accept方法
    func (ln tcpKeepAliveListener) Accept() (net.Conn, error) { 
    	return ln.AcceptTCP()
    }

# 三、TCP Conn的实现 #

package tcp
    
    type conn struct { 
    	server     *Server
    	cancelCtx  context.CancelFunc
    	rwc        net.Conn
    	remoteAddr string
    }
    
    func (c *conn) close() { 
    	c.rwc.Close()
    }
    
    func (c *conn) serve(ctx context.Context) { 
    	defer func() { 
    		if err := recover(); err != nil && err != ErrAbortHandler { 
    			const size = 64 << 10
    			buf := make([]byte, size)
    			buf = buf[:runtime.Stack(buf, false)]
    			fmt.Printf("tcp: panic serving %v: %v\n%s", c.remoteAddr, err, buf)
    		}
    		c.close()
    	}()
    	c.remoteAddr = c.rwc.RemoteAddr().String()
    	ctx = context.WithValue(ctx, LocalAddrContextKey, c.rwc.LocalAddr())
    	if c.server.Handler == nil { 
    		panic("invalid handler")
    	}
    	c.server.Handler.ServeTCP(ctx, c.rwc)
    }

# 四、Handler的实现 #

package main
    
    type handler struct{ }
    
    func (t *handler) ServeTCP(ctx context.Context, conn net.Conn) { 
    	conn.Write([]byte("tcp handler\n"))
    }
    
    func main() { 
    	addr := ":3001"
    	log.Println("start tcp-server at ", addr)
    	srv := &tcp.Server{ 
    		Addr:    addr,
    		Handler: &handler{ },
    	}
    	log.Fatal(srv.ListenAndServe())
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221021/177aca25020044cb980bbf74cd8cc725.png