7.1 Zinx-V0.7代码实现

我们的代码改动并不是很大。

A) 添加读写模块交互数据的管道

zinx/znet/connection.go

  1. type Connection struct {
  2. //当前连接的socket TCP套接字
  3. Conn *net.TCPConn
  4. //当前连接的ID 也可以称作为SessionID,ID全局唯一
  5. ConnID uint32
  6. //当前连接的关闭状态
  7. isClosed bool
  8. //消息管理MsgId和对应处理方法的消息管理模块
  9. MsgHandler ziface.IMsgHandle
  10. //告知该链接已经退出/停止的channel
  11. ExitBuffChan chan bool
  12. //无缓冲管道,用于读、写两个goroutine之间的消息通信
  13. msgChan chan []byte
  14. }
  15. //创建连接的方法
  16. func NewConntion(conn *net.TCPConn, connID uint32, msgHandler ziface.IMsgHandle) *Connection{
  17. c := &Connection{
  18. Conn: conn,
  19. ConnID: connID,
  20. isClosed: false,
  21. MsgHandler: msgHandler,
  22. ExitBuffChan: make(chan bool, 1),
  23. msgChan:make(chan []byte), //msgChan初始化
  24. }
  25. return c
  26. }

我们给Connection新增一个管道成员msgChan,作用是用于读写两个go的通信。

B) 创建Writer Goroutine

zinx/znet/connection.go

  1. /*
  2. 写消息Goroutine, 用户将数据发送给客户端
  3. */
  4. func (c *Connection) StartWriter() {
  5. fmt.Println("[Writer Goroutine is running]")
  6. defer fmt.Println(c.RemoteAddr().String(), "[conn Writer exit!]")
  7. for {
  8. select {
  9. case data := <-c.msgChan:
  10. //有数据要写给客户端
  11. if _, err := c.Conn.Write(data); err != nil {
  12. fmt.Println("Send Data error:, ", err, " Conn Writer exit")
  13. return
  14. }
  15. case <- c.ExitBuffChan:
  16. //conn已经关闭
  17. return
  18. }
  19. }
  20. }
C) Reader讲发送客户端的数据改为发送至Channel

修改Reader调用的SendMsg()方法

zinx/znet/connection.go

  1. //直接将Message数据发送数据给远程的TCP客户端
  2. func (c *Connection) SendMsg(msgId uint32, data []byte) error {
  3. if c.isClosed == true {
  4. return errors.New("Connection closed when send msg")
  5. }
  6. //将data封包,并且发送
  7. dp := NewDataPack()
  8. msg, err := dp.Pack(NewMsgPackage(msgId, data))
  9. if err != nil {
  10. fmt.Println("Pack error msg id = ", msgId)
  11. return errors.New("Pack error msg ")
  12. }
  13. //写回客户端
  14. c.msgChan <- msg //将之前直接回写给conn.Write的方法 改为 发送给Channel 供Writer读取
  15. return nil
  16. }

D) 启动Reader和Writer

zinx/znet/connection.go

  1. //启动连接,让当前连接开始工作
  2. func (c *Connection) Start() {
  3. //1 开启用户从客户端读取数据流程的Goroutine
  4. go c.StartReader()
  5. //2 开启用于写回客户端数据流程的Goroutine
  6. go c.StartWriter()
  7. for {
  8. select {
  9. case <- c.ExitBuffChan:
  10. //得到退出消息,不再阻塞
  11. return
  12. }
  13. }
  14. }