Go完整即时通讯项目及Go的生态介绍

末蓝、 2024-03-17 09:23 147阅读 0赞

Go完整即时通讯项目

项目架构：
在这里插入图片描述

1 编写基本服务端-Server

server.go

  1. package main
  3. import (
  4.     "fmt"
  5.     "net"
  6. )
  8. // 定义服务端
  9. type Server struct {
  11.     ip   string
  12.     port int
  13. }
  15. // 创建一个Server
  16. func NewServer(ip string, port int) *Server {
  18.     return &Server{
  20.         ip:   ip,
  21.         port: port,
  22.     }
  23. }
  25. // 定义Server方法
  26. func (this *Server) Start() {
  28.     //根据ip+port监听socket套接字 tcp表明类型【socket listen】
  29.     listen, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.ip, this.port))
  30.     if err != nil {
  32.         fmt.Println("net.Listen err=", err)
  33.         return
  34.     }
  35.     //关闭套接字，避免资源浪费
  36.     defer listen.Close()
  37.     for {
  39.         conn, err := listen.Accept()
  40.         if err != nil {
  42.             fmt.Println("listen accept err=", err)
  43.             continue
  44.         }
  45.         //处理连接请求【具体业务：处理客户端请求】
  46.         //开启协程处理，避免占用主线程【server一直要监听ip+port】
  47.         go this.Handler(conn)
  48.     }
  49. }
  51. // 处理连接请求
  52. func (this *Server) Handler(conn net.Conn) {
  54.     //当前连接的业务
  55.     fmt.Printf("连接建立成功...")
  56. }

通过在main.go中启动一个server并配合telnet命令检测代码是否正确

  • telnet:可以模拟连接的建立
  • telnet 127.0.0.1 8082

main.go:

  1. package main
  3. func main() {
  5.     //创建一个server
  6.     server := NewServer("127.0.0.1", 8082)
  7.     //启动server【监听】
  8.     server.Start()
  9. }
  11. //打包代码为exe
  12. go build -o intime.exe .\main.go .\server.go

在这里插入图片描述

2 实现用户上线广播机制【用户上线功能】

架构图：Server端存储一个OnlineMap，用于记录在线的用户

在这里插入图片描述

  1. 编写user.go，编写User结构体并实现对user.channel的监听
  2. 修改server.go，新增OnlineMap和Message属性，在处理的客户端上线的Handler中连接建立成功之后将用户添加到OnlineMap；并新增广播消息方法
  3. 在server.go中新增监听广播消息channel的方法，同时用一个goroutine单独监听message
  1. //构建代码 生成intime.exe文件
  2. go build -o intime.exe .\main.go .\server.go .\user.go

3 用户消息广播机制

修改server.go，完善一个handle处理业务方法，启动一个专门针对当前用户的goroutine

server.go:

  1. package main
  3. import (
  4.     "fmt"
  5.     "io"
  6.     "net"
  7.     "sync"
  8. )
  10. // 定义服务端
  11. type Server struct {
  13.     ip   string
  14.     port int
  16.     //定义一个map，用于存储在线用户
  17.     OnlineMap map[string]*User
  18.     //定义map锁，保证存储map时候的数据正确，避免并发读取
  19.     mapLock sync.RWMutex
  20.     //消息广播的channel[当该chan中有数据时，直接广播给所有在线用户]
  21.     Message chan string
  22. }
  24. // 创建一个Server
  25. func NewServer(ip string, port int) *Server {
  27.     return &Server{
  29.         ip:        ip,
  30.         port:      port,
  31.         OnlineMap: make(map[string]*User),
  32.         Message:   make(chan string),
  33.     }
  34. }
  36. // 定义Server方法
  37. func (this *Server) Start() {
  39.     //根据ip+port监听socket套接字 tcp表明类型【socket listen】
  40.     listen, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.ip, this.port))
  41.     if err != nil {
  43.         fmt.Println("net.Listen err=", err)
  44.         return
  45.     }
  46.     //关闭套接字，避免资源浪费
  47.     defer listen.Close()
  48.     //启动监听Message的goroutine
  49.     go this.ListenMessage()
  51.     for {
  53.         conn, err := listen.Accept()
  54.         if err != nil {
  56.             fmt.Println("listen accept err=", err)
  57.             continue
  58.         }
  59.         //处理连接请求【具体业务：处理客户端请求】
  60.         //开启协程处理，避免占用主线程【server一直要监听ip+port】
  61.         go this.Handler(conn)
  62.     }
  63. }
  65. // 处理连接请求
  66. func (this *Server) Handler(conn net.Conn) {
  68.     //当前连接的业务[用户上线之后，发送广播通知]
  69.     //fmt.Printf("连接建立成功...")
  70.     //1. 创建user，并将user添加到OnlineMap中，此处操作map（可能涉及并发，因此加锁保证安全性）
  71.     user := NewUser(conn)
  72.     this.mapLock.Lock()
  73.     this.OnlineMap[user.Name] = user
  74.     this.mapLock.Unlock()
  75.     //2. 将该用户上线消息进行广播
  76.     this.Broadcast(user, "已上线")
  78.     //3. 接收用户发送的消息，并广播
  79.     go func() {
  81.         buf := make([]byte, 4096)
  82.         //接收用户信息
  83.         n, err := conn.Read(buf)
  84.         if n == 0 {
  86.             this.Broadcast(user, "下线")
  87.             return
  88.         }
  89.         if err != nil && err != io.EOF {
  91.             fmt.Println("Conn read err:", err)
  92.             return
  93.         }
  94.         //提取用户消息（去除"\n"）
  95.         msg := string(buf[:n-1])
  96.         //广播消息
  97.         this.Broadcast(user, msg)
  98.     }()
  100.     //4. 阻塞当前handler
  101.     select {
  102.     }
  103. }
  105. // 广播消息
  106. func (this *Server) Broadcast(user *User, message string) {
  108.     sendMsg := "[" + user.Addr + "] " + user.Name + ":" + message
  109.     //将消息写入广播chan中
  110.     this.Message <- sendMsg
  111. }
  113. // 监听Message广播消息channel的goroutine，一旦有消息就广播发送给所有在线用户
  114. func (this *Server) ListenMessage() {
  116.     for {
  118.         msg := <-this.Message
  119.         //遍历OnlineMap，将消息广播给所有用户[设置并发操作map，加锁]
  120.         this.mapLock.Lock()
  121.         for _, cli := range this.OnlineMap {
  123.             //将广播消息写入用户的channel，等待用户监听读取
  124.             cli.C <- msg
  125.         }
  126.         this.mapLock.Unlock()
  127.     }
  128. }

4 用户业务层封装

修改user.go，新增对应方法：

  • user中新增一个Server属性
  • Online
  • Offline
  • DoMessage等

替换之前server.go中涉及到user的代码

user.go:

  1. package main
  3. import (
  4.     "net"
  5.     "unicode/utf8"
  6. )
  8. type User struct {
  10.     Name string
  11.     Addr string
  12.     //管道用于接收服务端的消息
  13.     C chan string
  14.     //与服务器端的连接
  15.     conn net.Conn
  17.     //对应连接的Server
  18.     server *Server
  19. }
  21. func NewUser(conn net.Conn, server *Server) *User {
  23.     userAddr := conn.RemoteAddr().String()
  24.     user := &User{
  26.         Name:   userAddr,
  27.         Addr:   userAddr,
  28.         C:      make(chan string),
  29.         conn:   conn,
  30.         server: server,
  31.     }
  33.     //创建一个user就应该监听自己chan管道中的消息，如果有就取出
  34.     go user.ListenMessage()
  35.     return user
  36. }
  38. // 监听当前user 的channel，一旦有消息，就直接发送给对应的客户端
  39. func (this *User) ListenMessage() {
  41.     for {
  43.         msg := <-this.C
  44.         //从user管道中读取消息，发送给user客户端
  45.         对中文进行处理
  46.         //msgByte := []byte(msg)
  47.         //bytes, err := simplifiedchinese.GB18030.NewDecoder().Bytes(msgByte)
  48.         //if err != nil {
  50.         //    fmt.Println("simplifiedchinese decoder err=", err)
  51.         //}
  52.         runes := []rune(msg + "\n")
  53.         bytes := make([]byte, len(runes)*4)
  54.         for _, v := range runes {
  56.             buf := make([]byte, 4)
  57.             size := utf8.EncodeRune(buf, v)
  58.             bytes = append(bytes, buf[:size]...)
  59.         }
  61.         //this.conn.Write([]byte(msg + "\n"))
  62.         this.conn.Write(bytes)
  63.     }
  64. }
  66. // Online：用户上线方法
  67. func (this *User) Online() {
  69.     //用户上线，将用户添加到OnlineMap中
  70.     this.server.mapLock.Lock()
  71.     this.server.OnlineMap[this.Name] = this
  72.     this.server.mapLock.Unlock()
  74.     //广播用户上线信息
  75.     this.server.Broadcast(this, "已上线")
  76. }
  78. // 用户下线业务
  79. func (this *User) Offline() {
  81.     this.server.mapLock.Lock()
  82.     //根据key删除对应值
  83.     delete(this.server.OnlineMap, this.Name)
  84.     this.server.mapLock.Unlock()
  85.     this.server.Broadcast(this, "下线")
  86. }
  88. // 用户处理消息的业务
  89. func (this *User) DoMessage(msg string) {
  91.     this.server.Broadcast(this, msg)
  92. }

server.go:

  1. package main
  3. import (
  4.     "fmt"
  5.     "io"
  6.     "net"
  7.     "sync"
  8. )
  10. // 定义服务端
  11. type Server struct {
  13.     ip   string
  14.     port int
  16.     //定义一个map，用于存储在线用户
  17.     OnlineMap map[string]*User
  18.     //定义map锁，保证存储map时候的数据正确，避免并发读取
  19.     mapLock sync.RWMutex
  20.     //消息广播的channel[当该chan中有数据时，直接广播给所有在线用户]
  21.     Message chan string
  22. }
  24. // 创建一个Server
  25. func NewServer(ip string, port int) *Server {
  27.     return &Server{
  29.         ip:        ip,
  30.         port:      port,
  31.         OnlineMap: make(map[string]*User),
  32.         Message:   make(chan string),
  33.     }
  34. }
  36. // 定义Server方法
  37. func (this *Server) Start() {
  39.     //根据ip+port监听socket套接字 tcp表明类型【socket listen】
  40.     listen, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.ip, this.port))
  41.     if err != nil {
  43.         fmt.Println("net.Listen err=", err)
  44.         return
  45.     }
  46.     //关闭套接字，避免资源浪费
  47.     defer listen.Close()
  48.     //启动监听Message的goroutine
  49.     go this.ListenMessage()
  51.     for {
  53.         conn, err := listen.Accept()
  54.         if err != nil {
  56.             fmt.Println("listen accept err=", err)
  57.             continue
  58.         }
  59.         //处理连接请求【具体业务：处理客户端请求】
  60.         //开启协程处理，避免占用主线程【server一直要监听ip+port】
  61.         go this.Handler(conn)
  62.     }
  63. }
  65. // 处理连接请求
  66. func (this *Server) Handler(conn net.Conn) {
  68.     //当前连接的业务[用户上线之后，发送广播通知]
  69.     //fmt.Printf("连接建立成功...")
  70.     // 创建user，并将user添加到OnlineMap中，此处操作map（可能涉及并发，因此加锁保证安全性）
  71.     user := NewUser(conn, this)
  72.     user.Online()
  74.     // 接收用户发送的消息，并广播
  75.     go func() {
  77.         buf := make([]byte, 4096)
  78.         //接收用户信息
  79.         n, err := conn.Read(buf)
  80.         if n == 0 {
  82.             //用户下线
  83.             user.Offline()
  84.             return
  85.         }
  86.         if err != nil && err != io.EOF {
  88.             fmt.Println("Conn read err:", err)
  89.             return
  90.         }
  91.         //提取用户消息（去除"\n"）
  92.         msg := string(buf[:n-1])
  93.         //用户针对msg进行消息处理
  94.         user.DoMessage(msg)
  95.     }()
  97.     //4. 阻塞当前handler
  98.     select {
  99.     }
  100. }
  102. // 广播消息
  103. func (this *Server) Broadcast(user *User, message string) {
  105.     sendMsg := "[" + user.Addr + "] " + user.Name + ":" + message
  106.     //将消息写入广播chan中
  107.     this.Message <- sendMsg
  108. }
  110. // 监听Message广播消息channel的goroutine，一旦有消息就广播发送给所有在线用户
  111. func (this *Server) ListenMessage() {
  113.     for {
  115.         msg := <-this.Message
  116.         //遍历OnlineMap，将消息广播给所有用户[设置并发操作map，加锁]
  117.         this.mapLock.Lock()
  118.         for _, cli := range this.OnlineMap {
  120.             //将广播消息写入用户的channel，等待用户监听读取
  121.             cli.C <- msg
  122.         }
  123.         this.mapLock.Unlock()
  124.     }
  125. }

5 用户查询功能

实现，用户在终端输入who，查看当前在线用户（修改user.go）

  • 添加SendMsg():给客户端发送消息
  • 新增判断“who”命令逻辑
  1. package main
  3. import (
  4.     "net"
  5.     "unicode/utf8"
  6. )
  8. type User struct {
  10.     Name string
  11.     Addr string
  12.     //管道用于接收服务端的消息
  13.     C chan string
  14.     //与服务器端的连接
  15.     conn net.Conn
  17.     //对应连接的Server
  18.     server *Server
  19. }
  21. func NewUser(conn net.Conn, server *Server) *User {
  23.     userAddr := conn.RemoteAddr().String()
  24.     user := &User{
  26.         Name:   userAddr,
  27.         Addr:   userAddr,
  28.         C:      make(chan string),
  29.         conn:   conn,
  30.         server: server,
  31.     }
  33.     //创建一个user就应该监听自己chan管道中的消息，如果有就取出
  34.     go user.ListenMessage()
  35.     return user
  36. }
  38. // 监听当前user 的channel，一旦有消息，就直接发送给对应的客户端
  39. func (this *User) ListenMessage() {
  41.     for {
  43.         msg := <-this.C
  44.         //从user管道中读取消息，发送给user客户端
  45.         对中文进行处理
  46.         //msgByte := []byte(msg)
  47.         //bytes, err := simplifiedchinese.GB18030.NewDecoder().Bytes(msgByte)
  48.         //if err != nil {
  50.         //    fmt.Println("simplifiedchinese decoder err=", err)
  51.         //}
  52.         runes := []rune(msg + "\n")
  53.         bytes := make([]byte, len(runes)*4)
  54.         for _, v := range runes {
  56.             buf := make([]byte, 4)
  57.             size := utf8.EncodeRune(buf, v)
  58.             bytes = append(bytes, buf[:size]...)
  59.         }
  61.         //this.conn.Write([]byte(msg + "\n"))
  62.         this.conn.Write(bytes)
  63.     }
  64. }
  66. // Online：用户上线方法
  67. func (this *User) Online() {
  69.     //用户上线，将用户添加到OnlineMap中
  70.     this.server.mapLock.Lock()
  71.     this.server.OnlineMap[this.Name] = this
  72.     this.server.mapLock.Unlock()
  74.     //广播用户上线信息
  75.     this.server.Broadcast(this, "已上线")
  76. }
  78. // 用户下线业务
  79. func (this *User) Offline() {
  81.     this.server.mapLock.Lock()
  82.     //根据key删除对应值
  83.     delete(this.server.OnlineMap, this.Name)
  84.     this.server.mapLock.Unlock()
  85.     this.server.Broadcast(this, "下线")
  86. }
  88. // 用户处理消息的业务
  89. func (this *User) DoMessage(msg string) {
  91.     //添加who命令逻辑:查询当前在线用户
  92.     if msg == "who" {
  94.         this.server.mapLock.Lock()
  95.         for _, user := range this.server.OnlineMap {
  97.             onlineMsg := "[" + user.Addr + "]" + user.Name + ":" + "online...\n"
  98.             this.SendMsg(onlineMsg)
  99.         }
  100.     }
  101.     this.server.Broadcast(this, msg)
  102. }
  104. // 给当前user的客户端发送消息
  105. func (this *User) SendMsg(msg string) {
  107.     this.conn.Write([]byte(msg))
  108. }

6 修改用户名

定义命令rename|zhangsan:将当前用户名修改为张三

  • 修改user.go:在DoMessage（）方法中判断命令是否为rename

user.go:

  1. package main
  3. import (
  4.     "net"
  5.     "strings"
  6.     "unicode/utf8"
  7. )
  9. type User struct {
  11.     Name string
  12.     Addr string
  13.     //管道用于接收服务端的消息
  14.     C chan string
  15.     //与服务器端的连接
  16.     conn net.Conn
  18.     //对应连接的Server
  19.     server *Server
  20. }
  22. func NewUser(conn net.Conn, server *Server) *User {
  24.     userAddr := conn.RemoteAddr().String()
  25.     user := &User{
  27.         Name:   userAddr,
  28.         Addr:   userAddr,
  29.         C:      make(chan string),
  30.         conn:   conn,
  31.         server: server,
  32.     }
  34.     //创建一个user就应该监听自己chan管道中的消息，如果有就取出
  35.     go user.ListenMessage()
  36.     return user
  37. }
  39. // 监听当前user 的channel，一旦有消息，就直接发送给对应的客户端
  40. func (this *User) ListenMessage() {
  42.     for {
  44.         msg := <-this.C
  45.         //从user管道中读取消息，发送给user客户端
  46.         对中文进行处理
  47.         //msgByte := []byte(msg)
  48.         //bytes, err := simplifiedchinese.GB18030.NewDecoder().Bytes(msgByte)
  49.         //if err != nil {
  51.         //    fmt.Println("simplifiedchinese decoder err=", err)
  52.         //}
  53.         runes := []rune(msg + "\n")
  54.         bytes := make([]byte, len(runes)*4)
  55.         for _, v := range runes {
  57.             buf := make([]byte, 4)
  58.             size := utf8.EncodeRune(buf, v)
  59.             bytes = append(bytes, buf[:size]...)
  60.         }
  62.         //this.conn.Write([]byte(msg + "\n"))
  63.         this.conn.Write(bytes)
  64.     }
  65. }
  67. // Online：用户上线方法
  68. func (this *User) Online() {
  70.     //用户上线，将用户添加到OnlineMap中
  71.     this.server.mapLock.Lock()
  72.     this.server.OnlineMap[this.Name] = this
  73.     this.server.mapLock.Unlock()
  75.     //广播用户上线信息
  76.     this.server.Broadcast(this, "is online")
  77. }
  79. // 用户下线业务
  80. func (this *User) Offline() {
  82.     this.server.mapLock.Lock()
  83.     //根据key删除对应值
  84.     delete(this.server.OnlineMap, this.Name)
  85.     this.server.mapLock.Unlock()
  86.     this.server.Broadcast(this, "is offline")
  87. }
  89. // 用户处理消息的业务
  90. func (this *User) DoMessage(msg string) {
  92.     //添加who命令逻辑:查询当前在线用户
  93.     if msg == "who" {
  95.         this.server.mapLock.Lock()
  96.         for _, user := range this.server.OnlineMap {
  98.             onlineMsg := "[" + user.Addr + "]" + user.Name + ":" + "online...\n"
  99.             this.SendMsg(onlineMsg)
  100.         }
  101.     } else if len(msg) > 7 && msg[:7] == "rename|" {
  103.         newName := strings.Split(msg, "|")[1]
  104.         //判断要修改的name是否已经被占用
  105.         _, ok := this.server.OnlineMap[newName]
  106.         if ok {
  108.             this.SendMsg("the name is already exists...")
  109.         } else {
  111.             this.server.mapLock.Lock()
  112.             delete(this.server.OnlineMap, this.Name)
  113.             this.server.OnlineMap[newName] = this
  114.             this.server.mapLock.Unlock()
  115.             this.Name = newName //更新页面当前用户
  116.             this.SendMsg("update name success:" + this.Name + "\n")
  117.         }
  118.     } else {
  120.         this.server.Broadcast(this, msg)
  121.     }
  122. }
  124. // 给当前user的客户端发送消息
  125. func (this *User) SendMsg(msg string) {
  127.     this.conn.Write([]byte(msg))
  128. }

7 超时强踢功能

如果某个用户长时间不发消息，不活跃，达到一定时间则断开连接，达到强踢效果

  • 修改server.go:
    ①在用户的Hander() goroutine中，添加用户活跃channel，一旦有消息就向该channel发送数据
    ②在用户的Hander()goroutine中，添加定时器功能，超时则强踢

server.go:

  1. package main
  3. import (
  4.     "fmt"
  5.     "io"
  6.     "net"
  7.     "sync"
  8.     "time"
  9. )
  11. // 定义服务端
  12. type Server struct {
  14.     ip   string
  15.     port int
  17.     //定义一个map，用于存储在线用户
  18.     OnlineMap map[string]*User
  19.     //定义map锁，保证存储map时候的数据正确，避免并发读取
  20.     mapLock sync.RWMutex
  21.     //消息广播的channel[当该chan中有数据时，直接广播给所有在线用户]
  22.     Message chan string
  23. }
  25. // 创建一个Server
  26. func NewServer(ip string, port int) *Server {
  28.     return &Server{
  30.         ip:        ip,
  31.         port:      port,
  32.         OnlineMap: make(map[string]*User),
  33.         Message:   make(chan string),
  34.     }
  35. }
  37. // 定义Server方法
  38. func (this *Server) Start() {
  40.     //根据ip+port监听socket套接字 tcp表明类型【socket listen】
  41.     listen, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.ip, this.port))
  42.     if err != nil {
  44.         fmt.Println("net.Listen err=", err)
  45.         return
  46.     }
  47.     //关闭套接字，避免资源浪费
  48.     defer listen.Close()
  49.     //启动监听Message的goroutine
  50.     go this.ListenMessage()
  52.     for {
  54.         conn, err := listen.Accept()
  55.         if err != nil {
  57.             fmt.Println("listen accept err=", err)
  58.             continue
  59.         }
  60.         //处理连接请求【具体业务：处理客户端请求】
  61.         //开启协程处理，避免占用主线程【server一直要监听ip+port】
  62.         go this.Handler(conn)
  63.     }
  64. }
  66. // 处理连接请求
  67. func (this *Server) Handler(conn net.Conn) {
  69.     //当前连接的业务[用户上线之后，发送广播通知]
  70.     //fmt.Printf("连接建立成功...")
  71.     // 创建user，并将user添加到OnlineMap中，此处操作map（可能涉及并发，因此加锁保证安全性）
  72.     user := NewUser(conn, this)
  73.     user.Online()
  74.     //监听用户是否活跃的channel
  75.     isLive := make(chan bool)
  77.     // 接收用户发送的消息，并广播
  78.     go func() {
  80.         buf := make([]byte, 4096)
  81.         //接收用户信息
  82.         n, err := conn.Read(buf)
  83.         if n == 0 {
  85.             //用户下线
  86.             user.Offline()
  87.             return
  88.         }
  89.         if err != nil && err != io.EOF {
  91.             fmt.Println("Conn read err:", err)
  92.             return
  93.         }
  94.         //提取用户消息（去除"\n"）
  95.         msg := string(buf[:n-1])
  96.         //用户针对msg进行消息处理
  97.         user.DoMessage(msg)
  98.         //用户的任意消息代表用户当前是一个活跃的
  99.         isLive <- true
  100.     }()
  102.     //4. 当前handler阻塞【超时强制踢出】
  103.     for {
  105.         select {
  107.         case <-isLive:
  108.             //当前用户是活跃的，应该重置定时器
  109.             //不做任何事，为了激活select，更新下面的定时器
  110.         case <-time.After(time.Second * 10):
  111.             //已经超时，将当前的User强制关闭
  112.             user.SendMsg("you have been offline")
  113.             //销毁用的资源,关闭channel
  114.             close(user.C)
  115.             //关闭连接
  116.             conn.Close()
  117.             //退出当前Handler[runtime.Goexit()]
  118.             return
  119.         }
  120.     }
  121. }
  123. // 广播消息
  124. func (this *Server) Broadcast(user *User, message string) {
  126.     sendMsg := "[" + user.Addr + "] " + user.Name + ":" + message
  127.     //将消息写入广播chan中
  128.     this.Message <- sendMsg
  129. }
  131. // 监听Message广播消息channel的goroutine，一旦有消息就广播发送给所有在线用户
  132. func (this *Server) ListenMessage() {
  134.     for {
  136.         msg := <-this.Message
  137.         //遍历OnlineMap，将消息广播给所有用户[设置并发操作map，加锁]
  138.         this.mapLock.Lock()
  139.         for _, cli := range this.OnlineMap {
  141.             //将广播消息写入用户的channel，等待用户监听读取
  142.             cli.C <- msg
  143.         }
  144.         this.mapLock.Unlock()
  145.     }
  146. }

8 私聊功能

消息格式：to|zhangsan|hello, how are you

  • 修改user.go的DoMessage()逻辑，新增私聊消息判断

user.go:

  1. package main
  3. import (
  4.     "net"
  5.     "strings"
  6.     "unicode/utf8"
  7. )
  9. type User struct {
  11.     Name string
  12.     Addr string
  13.     //管道用于接收服务端的消息
  14.     C chan string
  15.     //与服务器端的连接
  16.     conn net.Conn
  18.     //对应连接的Server
  19.     server *Server
  20. }
  22. func NewUser(conn net.Conn, server *Server) *User {
  24.     userAddr := conn.RemoteAddr().String()
  25.     user := &User{
  27.         Name:   userAddr,
  28.         Addr:   userAddr,
  29.         C:      make(chan string),
  30.         conn:   conn,
  31.         server: server,
  32.     }
  34.     //创建一个user就应该监听自己chan管道中的消息，如果有就取出
  35.     go user.ListenMessage()
  36.     return user
  37. }
  39. // 监听当前user 的channel，一旦有消息，就直接发送给对应的客户端
  40. func (this *User) ListenMessage() {
  42.     for {
  44.         msg := <-this.C
  45.         //从user管道中读取消息，发送给user客户端
  46.         对中文进行处理
  47.         //msgByte := []byte(msg)
  48.         //bytes, err := simplifiedchinese.GB18030.NewDecoder().Bytes(msgByte)
  49.         //if err != nil {
  51.         //    fmt.Println("simplifiedchinese decoder err=", err)
  52.         //}
  53.         runes := []rune(msg + "\n")
  54.         bytes := make([]byte, len(runes)*4)
  55.         for _, v := range runes {
  57.             buf := make([]byte, 4)
  58.             size := utf8.EncodeRune(buf, v)
  59.             bytes = append(bytes, buf[:size]...)
  60.         }
  62.         //this.conn.Write([]byte(msg + "\n"))
  63.         this.conn.Write(bytes)
  64.     }
  65. }
  67. // Online：用户上线方法
  68. func (this *User) Online() {
  70.     //用户上线，将用户添加到OnlineMap中
  71.     this.server.mapLock.Lock()
  72.     this.server.OnlineMap[this.Name] = this
  73.     this.server.mapLock.Unlock()
  75.     //广播用户上线信息
  76.     this.server.Broadcast(this, "is online")
  77. }
  79. // 用户下线业务
  80. func (this *User) Offline() {
  82.     this.server.mapLock.Lock()
  83.     //根据key删除对应值
  84.     delete(this.server.OnlineMap, this.Name)
  85.     this.server.mapLock.Unlock()
  86.     this.server.Broadcast(this, "is offline")
  87. }
  89. // 用户处理消息的业务
  90. func (this *User) DoMessage(msg string) {
  92.     //添加who命令逻辑:查询当前在线用户
  93.     if msg == "who" {
  95.         this.server.mapLock.Lock()
  96.         for _, user := range this.server.OnlineMap {
  98.             onlineMsg := "[" + user.Addr + "]" + user.Name + ":" + "online...\n"
  99.             this.SendMsg(onlineMsg)
  100.         }
  101.     } else if len(msg) > 7 && msg[:7] == "rename|" {
  103.         newName := strings.Split(msg, "|")[1]
  104.         //判断要修改的name是否已经被占用
  105.         _, ok := this.server.OnlineMap[newName]
  106.         if ok {
  108.             this.SendMsg("the name is already exists...")
  109.         } else {
  111.             this.server.mapLock.Lock()
  112.             delete(this.server.OnlineMap, this.Name)
  113.             this.server.OnlineMap[newName] = this
  114.             this.server.mapLock.Unlock()
  115.             this.Name = newName //更新页面当前用户
  116.             this.SendMsg("update name success:" + this.Name + "\n")
  117.         }
  118.     } else if len(msg) > 4 && msg[:3] == "to|" {
  120.         //如果是私聊命令 消息格式： to|zhangsan|msg content
  121.         //1. 获取对方用户名
  122.         remoteName := strings.Split(msg, "|")[1]
  123.         if remoteName == "" {
  125.             this.SendMsg("the msg format is incorrect, please use the 'to|zhangsan|msg content' to send a msg\n")
  126.             return
  127.         }
  128.         //2. 根据用户名，得到对方的user对象
  129.         remoteUser, ok := this.server.OnlineMap[remoteName]
  130.         if !ok {
  132.             this.SendMsg("the user is not exist")
  133.             return
  134.         }
  135.         //3. 获取消息内容，通过对方的User对象将消息内容发送过去
  136.         content := strings.Split(msg, "|")[2]
  137.         if content == "" {
  139.             this.SendMsg("please do not send a empty msg\n")
  140.             return
  141.         }
  142.         remoteUser.SendMsg(this.Name + "is speak to you:" + content)
  143.     } else {
  145.         this.server.Broadcast(this, msg)
  146.     }
  147. }
  149. // 给当前user的客户端发送消息
  150. func (this *User) SendMsg(msg string) {
  152.     this.conn.Write([]byte(msg))
  153. }

9 客户端实现（过程省略）

9.1 客户端类型定义与链接

9.2 解析命令行

9.3 菜单显示

9.4 更新用户名客户端实现

9.5 公聊模式

9.6 私聊模式

10 最终代码

①main.go

  1. package main
  3. func main() {
  5.     server := NewServer("127.0.0.1", 8888)
  6.     server.Start()
  7. }

②server.go

  1. package main
  3. import (
  4.     "fmt"
  5.     "io"
  6.     "net"
  7.     "sync"
  8.     "time"
  9. )
  11. type Server struct {
  13.     Ip   string
  14.     Port int
  16.     //在线用户的列表
  17.     OnlineMap map[string]*User
  18.     mapLock   sync.RWMutex
  20.     //消息广播的channel
  21.     Message chan string
  22. }
  24. //创建一个server的接口
  25. func NewServer(ip string, port int) *Server {
  27.     server := &Server{
  29.         Ip:        ip,
  30.         Port:      port,
  31.         OnlineMap: make(map[string]*User),
  32.         Message:   make(chan string),
  33.     }
  35.     return server
  36. }
  38. //监听Message广播消息channel的goroutine，一旦有消息就发送给全部的在线User
  39. func (this *Server) ListenMessager() {
  41.     for {
  43.         msg := <-this.Message
  45.         //将msg发送给全部的在线User
  46.         this.mapLock.Lock()
  47.         for _, cli := range this.OnlineMap {
  49.             cli.C <- msg
  50.         }
  51.         this.mapLock.Unlock()
  52.     }
  53. }
  55. //广播消息的方法
  56. func (this *Server) BroadCast(user *User, msg string) {
  58.     sendMsg := "[" + user.Addr + "]" + user.Name + ":" + msg
  60.     this.Message <- sendMsg
  61. }
  63. func (this *Server) Handler(conn net.Conn) {
  65.     //...当前链接的业务
  66.     //fmt.Println("链接建立成功")
  68.     user := NewUser(conn, this)
  70.     user.Online()
  72.     //监听用户是否活跃的channel
  73.     isLive := make(chan bool)
  75.     //接受客户端发送的消息
  76.     go func() {
  78.         buf := make([]byte, 4096)
  79.         for {
  81.             n, err := conn.Read(buf)
  82.             if n == 0 {
  84.                 user.Offline()
  85.                 return
  86.             }
  88.             if err != nil && err != io.EOF {
  90.                 fmt.Println("Conn Read err:", err)
  91.                 return
  92.             }
  94.             //提取用户的消息(去除'\n')
  95.             msg := string(buf[:n-1])
  97.             //用户针对msg进行消息处理
  98.             user.DoMessage(msg)
  100.             //用户的任意消息，代表当前用户是一个活跃的
  101.             isLive <- true
  102.         }
  103.     }()
  105.     //当前handler阻塞
  106.     for {
  108.         select {
  110.         case <-isLive:
  111.             //当前用户是活跃的，应该重置定时器
  112.             //不做任何事情，为了激活select，更新下面的定时器
  114.         case <-time.After(time.Second * 300):
  115.             //已经超时
  116.             //将当前的User强制的关闭
  118.             user.SendMsg("你被踢了")
  120.             //销毁用的资源
  121.             close(user.C)
  123.             //关闭连接
  124.             conn.Close()
  126.             //退出当前Handler
  127.             return //runtime.Goexit()
  128.         }
  129.     }
  130. }
  132. //启动服务器的接口
  133. func (this *Server) Start() {
  135.     //socket listen
  136.     listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.Ip, this.Port))
  137.     if err != nil {
  139.         fmt.Println("net.Listen err:", err)
  140.         return
  141.     }
  142.     //close listen socket
  143.     defer listener.Close()
  145.     //启动监听Message的goroutine
  146.     go this.ListenMessager()
  148.     for {
  150.         //accept
  151.         conn, err := listener.Accept()
  152.         if err != nil {
  154.             fmt.Println("listener accept err:", err)
  155.             continue
  156.         }
  158.         //do handler
  159.         go this.Handler(conn)
  160.     }
  161. }

③user.go

  1. package main
  3. import (
  4.     "net"
  5.     "strings"
  6. )
  8. type User struct {
  10.     Name string
  11.     Addr string
  12.     C    chan string
  13.     conn net.Conn
  15.     server *Server
  16. }
  18. //创建一个用户的API
  19. func NewUser(conn net.Conn, server *Server) *User {
  21.     userAddr := conn.RemoteAddr().String()
  23.     user := &User{
  25.         Name: userAddr,
  26.         Addr: userAddr,
  27.         C:    make(chan string),
  28.         conn: conn,
  30.         server: server,
  31.     }
  33.     //启动监听当前user channel消息的goroutine
  34.     go user.ListenMessage()
  36.     return user
  37. }
  39. //用户的上线业务
  40. func (this *User) Online() {
  43.     //用户上线,将用户加入到onlineMap中
  44.     this.server.mapLock.Lock()
  45.     this.server.OnlineMap[this.Name] = this
  46.     this.server.mapLock.Unlock()
  48.     //广播当前用户上线消息
  49.     this.server.BroadCast(this, "已上线")
  50. }
  52. //用户的下线业务
  53. func (this *User) Offline() {
  56.     //用户下线,将用户从onlineMap中删除
  57.     this.server.mapLock.Lock()
  58.     delete(this.server.OnlineMap, this.Name)
  59.     this.server.mapLock.Unlock()
  61.     //广播当前用户上线消息
  62.     this.server.BroadCast(this, "下线")
  64. }
  66. //给当前User对应的客户端发送消息
  67. func (this *User) SendMsg(msg string) {
  69.     this.conn.Write([]byte(msg))
  70. }
  72. //用户处理消息的业务
  73. func (this *User) DoMessage(msg string) {
  75.     if msg == "who" {
  77.         //查询当前在线用户都有哪些
  79.         this.server.mapLock.Lock()
  80.         for _, user := range this.server.OnlineMap {
  82.             onlineMsg := "[" + user.Addr + "]" + user.Name + ":" + "在线...\n"
  83.             this.SendMsg(onlineMsg)
  84.         }
  85.         this.server.mapLock.Unlock()
  87.     } else if len(msg) > 7 && msg[:7] == "rename|" {
  89.         //消息格式: rename|张三
  90.         newName := strings.Split(msg, "|")[1]
  92.         //判断name是否存在
  93.         _, ok := this.server.OnlineMap[newName]
  94.         if ok {
  96.             this.SendMsg("当前用户名被使用\n")
  97.         } else {
  99.             this.server.mapLock.Lock()
  100.             delete(this.server.OnlineMap, this.Name)
  101.             this.server.OnlineMap[newName] = this
  102.             this.server.mapLock.Unlock()
  104.             this.Name = newName
  105.             this.SendMsg("您已经更新用户名:" + this.Name + "\n")
  106.         }
  108.     } else if len(msg) > 4 && msg[:3] == "to|" {
  110.         //消息格式:  to|张三|消息内容
  112.         //1 获取对方的用户名
  113.         remoteName := strings.Split(msg, "|")[1]
  114.         if remoteName == "" {
  116.             this.SendMsg("消息格式不正确，请使用 \"to|张三|你好啊\"格式。\n")
  117.             return
  118.         }
  120.         //2 根据用户名 得到对方User对象
  121.         remoteUser, ok := this.server.OnlineMap[remoteName]
  122.         if !ok {
  124.             this.SendMsg("该用户名不不存在\n")
  125.             return
  126.         }
  128.         //3 获取消息内容，通过对方的User对象将消息内容发送过去
  129.         content := strings.Split(msg, "|")[2]
  130.         if content == "" {
  132.             this.SendMsg("无消息内容，请重发\n")
  133.             return
  134.         }
  135.         remoteUser.SendMsg(this.Name + "对您说:" + content)
  137.     } else {
  139.         this.server.BroadCast(this, msg)
  140.     }
  141. }
  143. //监听当前User channel的 方法,一旦有消息，就直接发送给对端客户端
  144. func (this *User) ListenMessage() {
  146.     for {
  148.         msg := <-this.C
  150.         this.conn.Write([]byte(msg + "\n"))
  151.     }
  152. }

④client.go

  1. package main
  3. import (
  4.     "flag"
  5.     "fmt"
  6.     "io"
  7.     "net"
  8.     "os"
  9. )
  11. type Client struct {
  13.     ServerIp   string
  14.     ServerPort int
  15.     Name       string
  16.     conn       net.Conn
  17.     flag       int //当前client的模式
  18. }
  20. func NewClient(serverIp string, serverPort int) *Client {
  22.     //创建客户端对象
  23.     client := &Client{
  25.         ServerIp:   serverIp,
  26.         ServerPort: serverPort,
  27.         flag:       999,
  28.     }
  30.     //链接server
  31.     conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", serverIp, serverPort))
  32.     if err != nil {
  34.         fmt.Println("net.Dial error:", err)
  35.         return nil
  36.     }
  38.     client.conn = conn
  40.     //返回对象
  41.     return client
  42. }
  44. //处理server回应的消息， 直接显示到标准输出即可
  45. func (client *Client) DealResponse() {
  47.     //一旦client.conn有数据，就直接copy到stdout标准输出上, 永久阻塞监听
  48.     io.Copy(os.Stdout, client.conn)
  49. }
  51. func (client *Client) menu() bool {
  53.     var flag int
  55.     fmt.Println("1.公聊模式")
  56.     fmt.Println("2.私聊模式")
  57.     fmt.Println("3.更新用户名")
  58.     fmt.Println("0.退出")
  60.     fmt.Scanln(&flag)
  62.     if flag >= 0 && flag <= 3 {
  64.         client.flag = flag
  65.         return true
  66.     } else {
  68.         fmt.Println(">>>>请输入合法范围内的数字<<<<")
  69.         return false
  70.     }
  72. }
  74. //查询在线用户
  75. func (client *Client) SelectUsers() {
  77.     sendMsg := "who\n"
  78.     _, err := client.conn.Write([]byte(sendMsg))
  79.     if err != nil {
  81.         fmt.Println("conn Write err:", err)
  82.         return
  83.     }
  84. }
  86. //私聊模式
  87. func (client *Client) PrivateChat() {
  89.     var remoteName string
  90.     var chatMsg string
  92.     client.SelectUsers()
  93.     fmt.Println(">>>>请输入聊天对象[用户名], exit退出:")
  94.     fmt.Scanln(&remoteName)
  96.     for remoteName != "exit" {
  98.         fmt.Println(">>>>请输入消息内容, exit退出:")
  99.         fmt.Scanln(&chatMsg)
  101.         for chatMsg != "exit" {
  103.             //消息不为空则发送
  104.             if len(chatMsg) != 0 {
  106.                 sendMsg := "to|" + remoteName + "|" + chatMsg + "\n\n"
  107.                 _, err := client.conn.Write([]byte(sendMsg))
  108.                 if err != nil {
  110.                     fmt.Println("conn Write err:", err)
  111.                     break
  112.                 }
  113.             }
  115.             chatMsg = ""
  116.             fmt.Println(">>>>请输入消息内容, exit退出:")
  117.             fmt.Scanln(&chatMsg)
  118.         }
  120.         client.SelectUsers()
  121.         fmt.Println(">>>>请输入聊天对象[用户名], exit退出:")
  122.         fmt.Scanln(&remoteName)
  123.     }
  124. }
  126. func (client *Client) PublicChat() {
  128.     //提示用户输入消息
  129.     var chatMsg string
  131.     fmt.Println(">>>>请输入聊天内容，exit退出.")
  132.     fmt.Scanln(&chatMsg)
  134.     for chatMsg != "exit" {
  136.         //发给服务器
  138.         //消息不为空则发送
  139.         if len(chatMsg) != 0 {
  141.             sendMsg := chatMsg + "\n"
  142.             _, err := client.conn.Write([]byte(sendMsg))
  143.             if err != nil {
  145.                 fmt.Println("conn Write err:", err)
  146.                 break
  147.             }
  148.         }
  150.         chatMsg = ""
  151.         fmt.Println(">>>>请输入聊天内容，exit退出.")
  152.         fmt.Scanln(&chatMsg)
  153.     }
  155. }
  157. func (client *Client) UpdateName() bool {
  160.     fmt.Println(">>>>请输入用户名:")
  161.     fmt.Scanln(&client.Name)
  163.     sendMsg := "rename|" + client.Name + "\n"
  164.     _, err := client.conn.Write([]byte(sendMsg))
  165.     if err != nil {
  167.         fmt.Println("conn.Write err:", err)
  168.         return false
  169.     }
  171.     return true
  172. }
  174. func (client *Client) Run() {
  176.     for client.flag != 0 {
  178.         for client.menu() != true {
  180.         }
  182.         //根据不同的模式处理不同的业务
  183.         switch client.flag {
  185.         case 1:
  186.             //公聊模式
  187.             client.PublicChat()
  188.             break
  189.         case 2:
  190.             //私聊模式
  191.             client.PrivateChat()
  192.             break
  193.         case 3:
  194.             //更新用户名
  195.             client.UpdateName()
  196.             break
  197.         }
  198.     }
  199. }
  201. var serverIp string
  202. var serverPort int
  204. //./client -ip 127.0.0.1 -port 8888
  205. func init() {
  207.     flag.StringVar(&serverIp, "ip", "127.0.0.1", "设置服务器IP地址(默认是127.0.0.1)")
  208.     flag.IntVar(&serverPort, "port", 8888, "设置服务器端口(默认是8888)")
  209. }
  211. func main() {
  213.     //命令行解析
  214.     flag.Parse()
  216.     client := NewClient(serverIp, serverPort)
  217.     if client == nil {
  219.         fmt.Println(">>>>> 链接服务器失败...")
  220.         return
  221.     }
  223.     //单独开启一个goroutine去处理server的回执消息
  224.     go client.DealResponse()
  226.     fmt.Println(">>>>>链接服务器成功...")
  228.     //启动客户端的业务
  229.     client.Run()
  230. }

11 go的全部生态

在这里插入图片描述

参考：

  • 资料地址：https://pan.baidu.com/s/1glckD7XGInHDFQQKCRE66g\#list/path=%2F

发表评论

表情：
评论列表 （有 0 条评论，147人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Go基础编程：Go语言介绍

    Go语言是什么 2009年11月10日，Go语言正式成为开源编程语言家庭的一员。 Go语言（或称Golang）是云计算时代的C语言。Go语言的诞生是为了让程序员有更高的

    桃扇骨桃扇骨/ 0 赞/ 471 阅读

    相关 go目录介绍

    一：工作区 所谓工作区可以理解为工作空间，由你安装go时指定工作区目录，比如我的：E:\\go\_project 此目录就是安装go时设置的GOPATH环境变量，GOPA

    柔光的暖阳◎柔光的暖阳◎/ 0 赞/ 479 阅读