EchoServer设计到实现（六）

问题

项目在经过思考后，重构了。主要重构原因有主要以下几点

1.包的设计粒度多大，导致依赖导致 2.代码gorouting操作不当，浪费cpu使用 3.网络模型使用不当，同步模型可以满足设计需求 4.select关键字使用不错，导致gorouting停止等待

重构后

修改协议设计，添加缓冲区，将网络和协议耦合降低。设计协议类，尽量以结构体为单位去做操作方法。 //思考：是否支持异步解析，现在支持同步解析 import ( "bytes" "encoding/binary" ) const ( //MsgLength 字符串长度 MsgLength = 4 ) //Protocol 协议 type Protocol struct { msgBuf []byte //ReaderChannel chan []byte } //New 返回新对象 func New() *Protocol { pro := &Protocol{msgBuf: make([]byte, 0)} //readerChannel:make(chan []byte,1024)} //go pro.Depack() return pro } //PushStream 添加缓冲区节流 func (pro *Protocol) PushStream(buf []byte) { pro.msgBuf = append(pro.msgBuf, buf...) } //Enpack 压缩 func (pro *Protocol) Enpack(message []byte) []byte { Header := IntToBytes(len(message)) return append(Header, message...) } //Depack 解压 func (pro *Protocol) Depack() []string { Msg := make([]string, 0) buffer := pro.msgBuf BufLen := len(pro.msgBuf) i := 0 for i < BufLen { if BufLen < i+MsgLength { break } msgLen := BytesToInt(buffer[i : i+MsgLength]) if msgLen > 0 && msgLen < 2<<32 { if BufLen < i+MsgLength+msgLen { break } data := buffer[i+MsgLength : i+MsgLength+msgLen] Msg = append(Msg, string(data)) i += MsgLength + msgLen } } if i == BufLen { pro.msgBuf = make([]byte, 0) } else { pro.msgBuf = buffer[i:] } return Msg } //IntToBytes 整形转换byte func IntToBytes(n int) []byte { x := int32(n) bytesBuffer := bytes.NewBuffer([]byte{}) binary.Write(bytesBuffer, binary.BigEndian, x) return bytesBuffer.Bytes() } //BytesToInt byte转换Int func BytesToInt(b []byte) int { bytesBuffer := bytes.NewBuffer(b) var x int32 binary.Read(bytesBuffer, binary.BigEndian, &x) return int(x) } 重构了网络操作，取代主函数，所有操作归并到网络类；更改接收和发送异步方式为同步操作。主要原因异步存在发送积压，导致客户端接收超时。 package server import ( "EchoTCP/config" "EchoTCP/logger" "EchoTCP/protocol" "net" "time" ) //Config 配置累 var Config = config.New(3) //EchoServer 服务sock结构 type EchoServer struct { Port string IP string Pro string } //Client 客户端 type Client struct { conn net.Conn tmpBuf protocol.Protocol } //New 创建新的EchoServer func New() *EchoServer { logger.Log.Logger.Info("Create EchoServer Sucesss") Config.Dyn.Monitor() return &EchoServer{} } //SetPort 设置端口号 func (server *EchoServer) SetPort(port string) { server.Port = port } //SetAddr 设置IP地址 func (server *EchoServer) SetAddr(addr string) { server.IP = addr } //SetPro 设置连接协议 func (server *EchoServer) SetPro(pro string) { server.Pro = pro } //Listen 启动监听服务 返回监听对象 错误址 func (server *EchoServer) Listen() (listener net.Listener, err error) { if server.IP == "" { server.IP = Config.Sta.Sock.IP } if server.Port == "" { server.Port = Config.Sta.Sock.Port } if "" == server.Pro { server.Pro = Config.Sta.Sock.Protocol } addr := server.IP + ":" + server.Port listener, err = net.Listen(server.Pro, addr) logger.Log.Logger.Info(addr) logger.Log.Logger.Info(server.Pro) if err != nil { logger.Log.Logger.Error(err.Error()) return nil, err } return listener, nil } //Accept 接受连接 func (server *EchoServer) Accept(listener net.Listener) { for { conn, err := listener.Accept() if err != nil { logger.Log.Logger.Error(err.Error()) continue } logger.Log.Logger.Info("There is Conner") client := Client{conn, *protocol.New()} go server.RecvHandle(client) } } //RecvHandle 接受客户端发送过的消息 func (server *EchoServer) RecvHandle(client Client) { buf := make([]byte, 1024) defer client.conn.Close() for { client.conn.SetReadDeadline(time.Now().Add(time.Second * 10)) n, err := client.conn.Read(buf) if err != nil { if nerr, ok := err.(net.Error); ok && nerr.Timeout() { logger.Log.Logger.Info("client timeout") } else { logger.Log.Logger.Error(err.Error()) } break } server.SendHandle(client, buf[:n]) } } //SendHandle 读写消息 func (server *EchoServer) SendHandle(client Client, buf []byte) { client.tmpBuf.PushStream(buf) recMsg := client.tmpBuf.Depack() for _, msg := range recMsg { logger.Log.Logger.Info(string("recv Msassge " + msg)) switch Config.Dyn.Opt.Select { case 1: msg = Reverse(msg) default: msg = Shuffle(msg) } _, err := server.SendSingleMsg(client.conn, msg) if err != nil { logger.Log.Logger.Error(err.Error()) } logger.Log.Logger.Info(string("send msg : " + msg)) } } //SendSingleMsg 单条发送 func (server *EchoServer) SendSingleMsg(conn net.Conn, sendMsg string) (n int, err error) { n, err = conn.Write([]byte(sendMsg)) return } 修改针对配置的设计，把动态配置和静态配置合并一起，去掉冗余代码，封装一个配置管理类，通过传入参数分类实力化动态配置还是静态配置。 type opteration struct { Select int } type dynConfig struct { Opt opteration Watcher *fsnotify.Watcher } type sock struct { Port string IP string Protocol string } type def struct { BufMax int Timeout int } type staConfig struct { Sock sock Def def } //Config 全局配置 type Config struct { Sta *staConfig Dyn *dynConfig } //New 生成新的对象 func New(configType int) (conf *Config) { switch configType { case 1: conf = &Config{Sta: newStaticConfig()} case 2: conf = &Config{Dyn: newDynConfig()} case 3: conf = &Config{Sta: newStaticConfig(), Dyn: newDynConfig()} } return } func newStaticConfig() *staConfig { var conf staConfig if err := readFileConfig(&conf, staticFilePath); err != nil { return nil } return &conf } func newDynConfig() *dynConfig { var conf dynConfig var err error if err = readFileConfig(&conf, dynFilePath); err != nil { return nil } conf.Watcher, err = fsnotify.NewWatcher() if err != nil { logger.Log.Logger.Error(err.Error()) return nil } err = conf.Watcher.Watch(dynFilePath) if err != nil { logger.Log.Logger.Error(err.Error()) return nil } return &conf } func readFileConfig(tlConfig interface{}, fileName string) (err error) { if _, err = toml.DecodeFile(fileName, tlConfig); err != nil { logger.Log.Logger.Error(err.Error()) return err } //fmt.Println(tlConfig, fileName) return nil } func (dyn *dynConfig) Monitor() { go func() { for { select { case <-dyn.Watcher.Event: temp := newDynConfig() if temp.Opt.Select != dyn.Opt.Select { dyn.Opt = temp.Opt } logger.Log.Logger.Info(string("Change : " + fmt.Sprintf("%d", temp.Opt.Select))) continue case err := <-dyn.Watcher.Error: logger.Log.Logger.Error(err.Error()) return default: time.Sleep(1 * time.Second) } } }() }

结束语

go语言的博大精深，设计理念由于没有具体的书籍很难学好。建议通过读go语言的源码，抛析源码学习使用go语言。 ps：项目暂时不给出，有需要可以私信我。