缘起

最近阅读<> (刘金亮, 2021.1)
本系列笔记拟采用golang练习之

案例需求(聊天服务器)

  • 用户可以连接到服务器。
  • 用户可以设定自己的用户名。
  • 用户可以向服务器发送消息，同时服务器也会向其他用户广播该消息。

目标

  • 实现聊天服务端, 支持端口监听, 多个客户端的连入, 消息收发, 广播, 断开, 并采集日志
  • 改造已有的聊天客户端, 使之能同时适配客户端和服务端的通信, 并设置写缓冲以防止死锁
  • 测试多个客户端的连入, 收发和断开, 并诊断服务端日志

设计

  • IMsg: 定义消息接口, 以及相关消息的实现. 为方便任意消息内容的解码, 消息传输时, 采用base64转码
  • IMsgDecoder: 定义消息解码器及其实现
  • IChatClient: 定义聊天客户端接口. 本次添加关闭通知方法, 以适配服务端.
  • tChatClient: 聊天客户端, 实现IChatClient接口. 本次添加关闭通知, 写缓冲和读超时控制, 修复写循环细节问题.
  • IChatServer: 定义聊天服务器接口, 为方便测试, 提供日志采集方法
  • tChatServer: 实现聊天服务器IChatServer

单元测试

ChatServer_test.go

  1. package chat_server
  3. import (
  4.     "fmt"
  5.     cs "learning/gooop/chat_server"
  6.     "strings"
  7.     "testing"
  8.     "time"
  9. )
  11. func Test_ChatServer(t *testing.T) {
  12.     fnAssertTrue := func(b bool, msg string) {
  13.         if !b {
  14.             t.Fatal(msg)
  15.         }
  16.     }
  18.     port := 3333
  19.     server := cs.NewChatServer()
  20.     err := server.Open(port)
  21.     if err != nil {
  22.         t.Fatal(err)
  23.     }
  25.     clientCount := 3
  26.     address := fmt.Sprintf("localhost:%v", port)
  27.     for i := 0;i < clientCount;i++ {
  28.         err, client := cs.DialChatClient(address)
  29.         if err != nil {
  30.             t.Fatal(err)
  31.         }
  33.         id := fmt.Sprintf("c%02d", i)
  34.         client.RecvHandler(func(client cs.IChatClient, msg cs.IMsg) {
  35.             t.Logf("%v recv: %v\n", id, msg)
  36.         })
  38.         go func() {
  39.             client.SetName(id)
  40.             client.Send(&cs.NameMsg{id })
  42.             n := 0
  43.             for range time.Tick(time.Duration(1) * time.Second) {
  44.                 client.Send(&cs.ChatMsg{id, fmt.Sprintf("msg %02d from %v", n, id) })
  46.                 n++
  47.                 if n >= 3 {
  48.                     break
  49.                 }
  50.             }
  52.             client.Close()
  53.         }()
  54.     }
  56.     passedSeconds := 0
  57.     for range time.Tick(time.Second) {
  58.         passedSeconds++
  59.         t.Logf("%v seconds passed", passedSeconds)
  61.         if passedSeconds >= 5 {
  62.             break
  63.         }
  64.     }
  65.     server.Close()
  67.     logs := server.GetLogs()
  68.     fnHasLog := func(log string) bool {
  69.         for _,it := range logs {
  70.             if strings.Contains(it, log) {
  71.                 return true
  72.             }
  73.         }
  74.         return false
  75.     }
  77.     for i := 0;i < clientCount;i++ {
  78.         msg := fmt.Sprintf("tChatServer.handleIncomingConn, clientCount=%v", i + 1)
  79.         fnAssertTrue(fnHasLog(msg), "expecting log: " + msg)
  81.         msg = fmt.Sprintf("tChatServer.handleClientClosed, c%02d", i)
  82.         fnAssertTrue(fnHasLog(msg), "expecting log: " + msg)
  83.     }
  84. }

测试输出

  1. $ go test -v ChatServer_test.go 
  2. === RUN   Test_ChatServer
  3. tChatServer.handleIncomingConn, clientCount=1
  4. tChatServer.handleIncomingConn, clientCount=2
  5. tChatServer.handleIncomingConn, clientCount=3
  6.     ChatServer_test.go:59: 1 seconds passed
  7.     ChatServer_test.go:35: c00 recv: &{c00 msg 00 from c00}
  8.     ChatServer_test.go:35: c02 recv: &{c00 msg 00 from c00}
  9.     ChatServer_test.go:35: c02 recv: &{c01 msg 00 from c01}
  10.     ChatServer_test.go:35: c01 recv: &{c00 msg 00 from c00}
  11.     ChatServer_test.go:35: c01 recv: &{c01 msg 00 from c01}
  12.     ChatServer_test.go:35: c01 recv: &{c02 msg 00 from c02}
  13.     ChatServer_test.go:35: c00 recv: &{c01 msg 00 from c01}
  14.     ChatServer_test.go:35: c00 recv: &{c02 msg 00 from c02}
  15.     ChatServer_test.go:35: c02 recv: &{c02 msg 00 from c02}
  16.     ChatServer_test.go:35: c00 recv: &{c01 msg 01 from c01}
  17.     ChatServer_test.go:35: c01 recv: &{c00 msg 01 from c00}
  18.     ChatServer_test.go:35: c01 recv: &{c02 msg 01 from c02}
  19.     ChatServer_test.go:35: c02 recv: &{c01 msg 01 from c01}
  20.     ChatServer_test.go:35: c02 recv: &{c00 msg 01 from c00}
  21.     ChatServer_test.go:59: 2 seconds passed
  22.     ChatServer_test.go:35: c00 recv: &{c00 msg 01 from c00}
  23.     ChatServer_test.go:35: c02 recv: &{c02 msg 01 from c02}
  24.     ChatServer_test.go:35: c00 recv: &{c02 msg 01 from c02}
  25. tChatClient.postConnClosed, c00, serverFlag=false
  26. tChatClient.postConnClosed, c02, serverFlag=false
  27. tChatClient.postConnClosed, c01, serverFlag=false
  28. tChatClient.postConnClosed, c02, serverFlag=true
  29. tChatServer.handleClientClosed, c02
  30. tChatServer.handleClientClosed, c02, clientCount=2
  31. tChatClient.postConnClosed, c01, serverFlag=true
  32. tChatServer.handleClientClosed, c01
  33. tChatServer.handleClientClosed, c01, clientCount=1
  34.     ChatServer_test.go:59: 3 seconds passed
  35. tChatClient.postConnClosed, c00, serverFlag=true
  36. tChatServer.handleClientClosed, c00
  37. tChatServer.handleClientClosed, c00, clientCount=0
  38.     ChatServer_test.go:59: 4 seconds passed
  39.     ChatServer_test.go:59: 5 seconds passed
  40. --- PASS: Test_ChatServer (5.00s)
  41. PASS
  42. ok      command-line-arguments  5.003s

IMsg.go

定义消息接口, 以及相关消息的实现. 为方便任意消息内容的解码, 消息传输时, 采用base64转码

  1. package chat_server
  3. import (
  4.     "encoding/base64"
  5.     "fmt"
  6. )
  8. type IMsg interface {
  9.     Encode() string
  10. }
  12. type NameMsg struct {
  13.     Name string
  14. }
  16. func (me *NameMsg) Encode() string {
  17.     return fmt.Sprintf("NAME %s\n", base64.StdEncoding.EncodeToString([]byte(me.Name)))
  18. }
  20. type ChatMsg struct {
  21.     Name string
  22.     Words string
  23. }
  25. func (me *ChatMsg) Encode() string {
  26.     return fmt.Sprintf("CHAT %s %s\n",
  27.         base64.StdEncoding.EncodeToString([]byte(me.Name)),
  28.         base64.StdEncoding.EncodeToString([]byte(me.Words)),
  29.     )
  30. }

IMsgDecoder.go

定义消息解码器及其实现

  1. package chat_server
  3. import (
  4.     "encoding/base64"
  5.     "strings"
  6. )
  9. type IMsgDecoder interface {
  10.     Decode(line string) (bool, IMsg)
  11. }
  13. type tMsgDecoder struct {
  14. }
  16. func (me *tMsgDecoder) Decode(line string) (bool, IMsg) {
  17.     items := strings.Split(line, " ")
  18.     size := len(items)
  20.     if items[0] == "NAME" && size == 2 {
  21.         name, err := base64.StdEncoding.DecodeString(items[1])
  22.         if err != nil {
  23.             return false, nil
  24.         }
  26.         return true, &NameMsg{
  27.             Name: string(name),
  28.         }
  29.     }
  31.     if items[0] == "CHAT" && size == 3 {
  32.         name, err := base64.StdEncoding.DecodeString(items[1])
  33.         if err != nil {
  34.             return false, nil
  35.         }
  37.         words, err := base64.StdEncoding.DecodeString(items[2])
  38.         if err != nil {
  39.             return false, nil
  40.         }
  42.         return true, &ChatMsg{
  43.             Name: string(name),
  44.             Words: string(words),
  45.         }
  46.     }
  48.     return false, nil
  49. }
  52. var MsgDecoder = &tMsgDecoder{}

IChatClient.go

定义聊天客户端接口. 本次添加关闭通知方法, 以适配服务端.

  1. package chat_server
  3. type IChatClient interface {
  4.     GetName() string
  5.     SetName(name string)
  7.     Send(msg IMsg)
  8.     RecvHandler(handler ClientRecvFunc)
  9.     CloseHandler(handler ClientCloseFunc)
  11.     Close()
  12. }
  14. type ClientRecvFunc func(client IChatClient, msg IMsg)
  15. type ClientCloseFunc func(client IChatClient)

tChatClient.go

聊天客户端, 实现IChatClient接口. 本次添加关闭通知, 写缓冲和读超时控制, 修复写循环细节问题.

  1. package chat_server
  3. import (
  4.     "bufio"
  5.     "fmt"
  6.     "io"
  7.     "net"
  8.     "sync/atomic"
  9.     "time"
  10. )
  12. type tChatClient struct {
  13.     conn net.Conn
  14.     name string
  15.     openFlag int32
  16.     closeFlag int32
  17.     serverFlag bool
  19.     closeChan chan bool
  20.     sendChan chan IMsg
  22.     sendLogs []IMsg
  23.     dropLogs []IMsg
  24.     recvLogs []IMsg
  25.     pendingSend int32
  27.     recvHandler ClientRecvFunc
  28.     closeHandler ClientCloseFunc
  29. }
  31. var gMaxPendingSend int32 = 100
  33. func DialChatClient(address string) (error, IChatClient) {
  34.     conn, err := net.Dial("tcp", address)
  35.     if err != nil {
  36.         return err, nil
  37.     }
  39.     return nil, openChatClient(conn, false)
  40. }
  42. func openChatClient(conn net.Conn, serverFlag bool) IChatClient {
  43.     it := &tChatClient{
  44.         conn: conn,
  45.         openFlag: 0,
  46.         closeFlag: 0,
  47.         serverFlag: serverFlag,
  49.         closeChan: make(chan bool),
  50.         sendChan: make(chan IMsg, gMaxPendingSend),
  52.         name: "anonymous",
  53.         sendLogs: []IMsg{},
  54.         dropLogs: []IMsg{},
  55.         recvLogs: []IMsg{},
  56.     }
  57.     it.open()
  58.     return it
  59. }
  62. func (me *tChatClient) GetName() string {
  63.     return me.name
  64. }
  66. func (me *tChatClient) SetName(name string) {
  67.     me.name = name
  68. }
  70. func (me *tChatClient) open(){
  71.     if !atomic.CompareAndSwapInt32(&me.openFlag, 0, 1) {
  72.         return
  73.     }
  75.     go me.beginWrite()
  76.     go me.beginRead()
  77. }
  80. func (me *tChatClient) isClosed() bool {
  81.     return me.closeFlag != 0
  82. }
  84. func (me *tChatClient) isNotClosed() bool {
  85.     return !me.isClosed()
  86. }
  88. func (me *tChatClient) Send(msg IMsg) {
  89.     if me.isClosed() {
  90.         return
  91.     }
  93.     if me.pendingSend < gMaxPendingSend {
  94.         atomic.AddInt32(&me.pendingSend, 1)
  95.         me.sendChan <- msg
  97.     } else {
  98.         me.dropLogs = append(me.dropLogs, msg)
  99.     }
  100. }
  102. func (me *tChatClient) RecvHandler(handler ClientRecvFunc) {
  103.     if me.isNotClosed() {
  104.         me.recvHandler = handler
  105.     }
  106. }
  109. func (me *tChatClient) CloseHandler(handler ClientCloseFunc) {
  110.     if me.isNotClosed() {
  111.         me.closeHandler = handler
  112.     }
  113. }
  116. func (me *tChatClient) Close() {
  117.     if me.isNotClosed() {
  118.         me.closeConn()
  119.     }
  120. }
  122. func (me *tChatClient) beginWrite() {
  123.     writer := io.Writer(me.conn)
  124.     for {
  125.         select {
  126.         case <- me.closeChan:
  127.             _ = me.conn.Close()
  128.             me.closeFlag = 2
  129.             me.postConnClosed()
  130.             return
  132.         case msg := <- me.sendChan:
  133.             atomic.AddInt32(&me.pendingSend, -1)
  134.             _,e := writer.Write([]byte(msg.Encode()))
  135.             if e != nil {
  136.                 me.closeConn()
  137.                 break
  138.             } else {
  139.                 me.sendLogs = append(me.sendLogs, msg)
  140.             }
  142.         case <- time.After(time.Duration(10) * time.Second):
  143.             me.postRecvTimeout()
  144.             break
  145.         }
  146.     }
  147. }
  149. func (me *tChatClient) postRecvTimeout() {
  150.     fmt.Printf("tChatClient.postRecvTimeout, %v, serverFlag=%v\n", me.name, me.serverFlag)
  151.     me.closeConn()
  152. }
  154. func (me *tChatClient) beginRead() {
  155.     reader := bufio.NewReader(me.conn)
  156.     for {
  157.         line, err := reader.ReadString('\n')
  158.         if err != nil {
  159.             me.closeConn()
  160.             break
  161.         }
  163.         ok, msg := MsgDecoder.Decode(line)
  164.         if ok {
  165.             fn := me.recvHandler
  166.             if fn != nil {
  167.                 fn(me, msg)
  168.             }
  170.             me.recvLogs = append(me.recvLogs, msg)
  171.         }
  172.     }
  173. }
  175. func (me *tChatClient) closeConn() {
  176.     if !atomic.CompareAndSwapInt32(&me.closeFlag, 0, 1) {
  177.         return
  178.     }
  179.     me.closeChan <- true
  180. }
  182. func (me *tChatClient) postConnClosed() {
  183.     fmt.Printf("tChatClient.postConnClosed, %v, serverFlag=%v\n", me.name, me.serverFlag)
  185.     handler := me.closeHandler
  186.     if handler != nil {
  187.         handler(me)
  188.     }
  190.     me.closeHandler = nil
  191.     me.recvHandler = nil
  192. }

IChatServer.go

定义聊天服务器接口, 为方便测试, 提供日志采集方法

  1. package chat_server
  3. type IChatServer interface {
  4.     Open(port int) error
  5.     Broadcast(msg IMsg)
  6.     Close()
  7.     GetLogs() []string
  8. }

tChatServer.go

实现聊天服务器IChatServer

  1. package chat_server
  3. import (
  4.     "errors"
  5.     "fmt"
  6.     "net"
  7.     "sync"
  8.     "sync/atomic"
  9. )
  11. type tChatServer struct {
  12.     openFlag int32
  13.     closeFlag int32
  15.     clients []IChatClient
  16.     clientCount int
  17.     clientLock *sync.RWMutex
  19.     listener net.Listener
  20.     recvLogs []IMsg
  22.     logs []string
  23. }
  25. func NewChatServer() IChatServer {
  26.     it := &tChatServer{
  27.         openFlag: 0,
  28.         closeFlag: 0,
  30.         clients: []IChatClient{},
  31.         clientCount: 0,
  32.         clientLock: new(sync.RWMutex),
  34.         listener: nil,
  35.         recvLogs: []IMsg{},
  36.     }
  37.     return it
  38. }
  40. func (me *tChatServer) Open(port int) error {
  41.     if !atomic.CompareAndSwapInt32(&me.openFlag, 0, 1) {
  42.         return errors.New("server already opened")
  43.     }
  45.     listener, err := net.Listen("tcp", fmt.Sprintf(":%v", port))
  46.     if err != nil {
  47.         return err
  48.     }
  50.     me.listener = listener
  51.     go me.beginListening()
  52.     return nil
  53. }
  55. func (me *tChatServer) logf(f string, args... interface{}) {
  56.     msg := fmt.Sprintf(f, args...)
  57.     me.logs = append(me.logs, msg)
  58.     fmt.Println(msg)
  59. }
  61. func (me *tChatServer) GetLogs() []string {
  62.     return me.logs
  63. }
  65. func (me *tChatServer) isClosed() bool {
  66.     return me.closeFlag != 0
  67. }
  69. func (me *tChatServer) isNotClosed() bool {
  70.     return !me.isClosed()
  71. }
  73. func (me *tChatServer) beginListening() {
  74.     for !me.isClosed() {
  75.         conn, err := me.listener.Accept()
  76.         if err != nil {
  77.             me.Close()
  78.             break
  79.         }
  81.         me.handleIncomingConn(conn)
  82.     }
  83. }
  86. func (me *tChatServer) Close() {
  87.     if !atomic.CompareAndSwapInt32(&me.closeFlag, 0, 1) {
  88.         return
  89.     }
  91.     _ = me.listener.Close()
  92.     me.closeAllClients()
  93. }
  95. func (me *tChatServer) closeAllClients() {
  96.     me.clientLock.Lock()
  97.     defer me.clientLock.Unlock()
  99.     for i,it := range me.clients {
  100.         if it != nil {
  101.             it.Close()
  102.             me.clients[i] = nil
  103.         }
  104.     }
  105.     me.clientCount = 0
  106. }
  109. func (me *tChatServer) handleIncomingConn(conn net.Conn) {
  110.     // init client
  111.     client := openChatClient(conn, true)
  112.     client.RecvHandler(me.handleClientMsg)
  113.     client.CloseHandler(me.handleClientClosed)
  115.     // lock me.clients
  116.     me.clientLock.Lock()
  117.     defer me.clientLock.Unlock()
  119.     // append to me.clients
  120.     if len(me.clients) > me.clientCount {
  121.         me.clients[me.clientCount] = client
  122.     } else {
  123.         me.clients = append(me.clients, client)
  124.     }
  125.     me.clientCount++
  127.     me.logf("tChatServer.handleIncomingConn, clientCount=%v", me.clientCount)
  128. }
  130. func (me *tChatServer) handleClientMsg(client IChatClient, msg IMsg) {
  131.     me.recvLogs = append(me.recvLogs, msg)
  133.     if nameMsg,ok := msg.(*NameMsg);ok {
  134.         client.SetName(nameMsg.Name)
  136.     } else if _, ok := msg.(*ChatMsg);ok {
  137.         me.Broadcast(msg)
  138.     }
  139. }
  141. func (me *tChatServer) handleClientClosed(client IChatClient) {
  142.     me.logf("tChatServer.handleClientClosed, %s", client.GetName())
  144.     me.clientLock.Lock()
  145.     defer me.clientLock.Unlock()
  147.     if me.clientCount <= 0 {
  148.         return
  149.     }
  151.     lastI := me.clientCount - 1
  152.     for i,it := range me.clients {
  153.         if it == client {
  154.             if i == lastI {
  155.                 me.clients[i] = nil
  156.             } else {
  157.                 me.clients[i], me.clients[lastI] = me.clients[lastI], nil
  158.             }
  159.             me.clientCount--
  160.             break
  161.         }
  162.     }
  164.     me.logf("tChatServer.handleClientClosed, %s, clientCount=%v", client.GetName(), me.clientCount)
  165. }
  167. func (me *tChatServer) Broadcast(msg IMsg) {
  168.     me.clientLock.RLock()
  169.     defer me.clientLock.RUnlock()
  171.     for _,it := range me.clients {
  172.         if it != nil {
  173.             it.Send(msg)
  174.         }
  175.     }
  176. }

(未完待续)