go 实现多人聊天系统服务端代码实现

向右看齐 2021-07-24 19:33 392阅读 0赞

# 一 代码结构图 #

![20210523162429999.png][]

## 二 model 包 ##

**chatroom\\server\\model\\error.go**

package model
    
    import (
       "errors"
    )
    
    // 根据业务逻辑需要，自定义一些错误。
    var (
       ERROR_USER_NOTEXISTS = errors.New("用户不存在")
       ERROR_USER_EXISTS    = errors.New("用户已经存在")
       ERROR_USER_PWD       = errors.New("密码不正确")
    )

**chatroom\\server\\model\\user.go**

package model
    
    // 定义一个用户的结构体
    type User struct {
       // 为了序列化和反序列化成功，必须保证用户信息的 json 字符串的 key 和 结构体的字段对应的 tag 名字一致
       UserId   int    `json:"userId"`
       UserPwd  string `json:"userPwd"`
       UserName string `json:"userName"`
    }

**chatroom\\server\\model\\userDao.go**

package model
    
    
    import (
       "chatroom/common/message"
       "encoding/json"
       "fmt"
       "github.com/garyburd/redigo/redis"
    )
    
    // 在服务器启动后，就初始化一个 userDao 实例。把它做成全局的变量，在需要在 redis 操作时，直接使用即可。
    var (
       MyUserDao *UserDao
    )
    
    // UserDao 结构体，完成对 User 结构体的各种操作。
    type UserDao struct {
       pool *redis.Pool
    }
    
    // 使用工厂模式，创建一个 UserDao 实例
    func NewUserDao(pool *redis.Pool) (userDao *UserDao) {
       userDao = &UserDao{
          pool: pool,
       }
       return
    }
    
    // 根据用户 id 返回一个 User实例 + err
    func (this *UserDao) getUserById(conn redis.Conn, id int) (user *User, err error) {
       // 通过给定 id 去 redis查询这个用户
       res, err := redis.String(conn.Do("HGet", "users", id))
       if err != nil {
          // 错误
          if err == redis.ErrNil { // 表示在 users 哈希中，没有找到对应id
             err = ERROR_USER_NOTEXISTS
          }
          return
       }
       user = &User{}
       // 需要把 res 反序列化成User实例
       err = json.Unmarshal([]byte(res), user)
       if err != nil {
          fmt.Println("json.Unmarshal err=", err)
          return
       }
       return
    }
    
    // 完成登录的校验
    // 1 Login 完成对用户的验证
    // 2 如果用户的 id 和 pwd 都正确，则返回一个 user 实例
    // 3 如果用户的 id 或 pwd 有错误，则返回对应的错误信息
    func (this *UserDao) Login(userId int, userPwd string) (user *User, err error) {
       // 先从 UserDao 的连接池中取出一根连接
       conn := this.pool.Get()
       defer conn.Close()
       user, err = this.getUserById(conn, userId)
       if err != nil {
          return
       }
       // 获取到用户后，对密码进行检查
       if user.UserPwd != userPwd {
          err = ERROR_USER_PWD
          return
       }
       return
    }
    
    // 注册用户
    func (this *UserDao) Register(user *message.User) (err error) {
       // 先从 UserDao 的连接池中取出一条连接
       conn := this.pool.Get()
       defer conn.Close()
       _, err = this.getUserById(conn, user.UserId)
       if err == nil {
          err = ERROR_USER_EXISTS
          return
       }
       // 说明 id 在 redis 还没有，则可以完成注册
       data, err := json.Marshal(user) // 序列化
       if err != nil {
          return
       }
       // 入库
       _, err = conn.Do("HSet", "users", user.UserId, string(data))
       if err != nil {
          fmt.Println("保存注册用户错误 err=", err)
          return
       }
       return
    }

# 三 utils 包 #

**chatroom\\server\\utils\\utils.go**

package utils
    
    import (
       "chatroom/common/message"
       "encoding/binary"
       "encoding/json"
       "fmt"
       "net"
    )
    
    // 将这些方法关联到 Transfer 结构体中
    type Transfer struct {
       // 连接
       Conn net.Conn
       // 传输数据的缓冲
       Buf [8096]byte
    }
    
    // 读消息
    func (this *Transfer) ReadPkg() (mes message.Message, err error) {
       fmt.Println("读取客户端发送的数据...")
       // conn.Read 在 conn 没有被关闭的情况下，才会阻塞
       // 如果客户端关闭了 conn 则，就不会阻塞
       _, err = this.Conn.Read(this.Buf[:4])
       if err != nil {
          return
       }
       // 根据 buf[:4] 转成一个 uint32 类型
       var pkgLen uint32
       pkgLen = binary.BigEndian.Uint32(this.Buf[0:4])
       // 根据 pkgLen 读取消息内容
       n, err := this.Conn.Read(this.Buf[:pkgLen])
       if n != int(pkgLen) || err != nil {
          return
       }
       // 反序列化成 message.Message
       err = json.Unmarshal(this.Buf[:pkgLen], &mes)
       if err != nil {
          fmt.Println("json.Unmarsha err=", err)
          return
       }
       return
    }
    
    // 写消息
    func (this *Transfer) WritePkg(data []byte) (err error) {
       // 先发送一个长度给对方
       var pkgLen uint32
       pkgLen = uint32(len(data))
       binary.BigEndian.PutUint32(this.Buf[0:4], pkgLen)
       // 发送长度
       n, err := this.Conn.Write(this.Buf[:4])
       if n != 4 || err != nil {
          fmt.Println("conn.Write(bytes) fail", err)
          return
       }
       // 发送 data 本身
       n, err = this.Conn.Write(data)
       if n != int(pkgLen) || err != nil {
          fmt.Println("conn.Write(bytes) fail", err)
          return
       }
       return
    }

# 四 process2 包 #

**chatroom\\server\\process\\smsProcess.go**

package process2
    
    import (
       "chatroom/common/message"
       "chatroom/server/utils"
       "encoding/json"
       "fmt"
       "net"
    )
    
    
    type SmsProcess struct {
       // 暂时不需要字段
    }
    
    
    // 群聊消息转发
    func (this *SmsProcess) SendGroupMes(mes *message.Message) {
       // 取出 mes 的内容 SmsMes
       var smsMes message.SmsMes
       err := json.Unmarshal([]byte(mes.Data), &smsMes)
       if err != nil {
          fmt.Println("json.Unmarshal err=", err)
          return
       }
       data, err := json.Marshal(mes)
       if err != nil {
          fmt.Println("json.Marshal err=", err)
          return
       }
       // 遍历服务器端的 onlineUsers map[int]*UserProcess,将消息转发出去。
       for id, up := range userMgr.onlineUsers {
          // 过滤掉自己,即不要再发给自己
          if id == smsMes.UserId {
             continue
          }
          this.SendMesToEachOnlineUser(data, up.Conn)
       }
    }
    func (this *SmsProcess) SendMesToEachOnlineUser(data []byte, conn net.Conn) {
       // 创建一个 Transfer 实例，发送 data
       tf := &utils.Transfer{
          Conn: conn,
       }
       err := tf.WritePkg(data)
       if err != nil {
          fmt.Println("转发消息失败 err=", err)
       }
    }

**chatroom\\server\\process\\userMgr.go**

package process2
    
    import (
       "fmt"
    )
    
    // 因为 UserMgr 实例在服务器端有且只有一个，并且很多的地方，都会使用到，所以将其定义为全局变量
    var (
       userMgr *UserMgr
    )
    
    type UserMgr struct {
       onlineUsers map[int]*UserProcess
    }
    
    // 完成对 userMgr 初始化工作
    func init() {
       userMgr = &UserMgr{
          onlineUsers: make(map[int]*UserProcess, 1024),
       }
    }
    
    // 完成对 onlineUsers 添加
    func (this *UserMgr) AddOnlineUser(up *UserProcess) {
       this.onlineUsers[up.UserId] = up
    }
    
    // 删除
    func (this *UserMgr) DelOnlineUser(userId int) {
       delete(this.onlineUsers, userId)
    }
    
    // 返回当前所有在线的用户
    func (this *UserMgr) GetAllOnlineUser() map[int]*UserProcess {
       return this.onlineUsers
    }
    
    // 根据 id 返回对应的用户
    func (this *UserMgr) GetOnlineUserById(userId int) (up *UserProcess, err error) {
       // 从 map 取出一个值，带检测方式
       up, ok := this.onlineUsers[userId]
       if !ok { // 要查找的这个用户，当前不在线。
          err = fmt.Errorf("用户%d 不存在", userId)
          return
       }
       return
    }

**chatroom\\server\\process\\userProcess.go**

package process2
    
    import (
       "chatroom/common/message"
       "chatroom/server/model"
       "chatroom/server/utils"
       "encoding/json"
       "fmt"
       "net"
    )
    
    type UserProcess struct {
       // 连接
       Conn net.Conn
       // 该 Conn 是哪个用户
       UserId int
    }
    
    // 通知所有在线的用户的方法
    // userId：通知其它的在线用户，我上线
    func (this *UserProcess) NotifyOthersOnlineUser(userId int) {
       // 遍历 onlineUsers , 然后一个一个的发送 NotifyUserStatusMes
       for id, up := range userMgr.onlineUsers {
          // 过滤掉自己
          if id == userId {
             continue
          }
          // 开始通知
          up.NotifyMeOnline(userId)
       }
    }
    
    // 通知用户上线消息
    func (this *UserProcess) NotifyMeOnline(userId int) {
       // 组装 NotifyUserStatusMes
       var mes message.Message
       mes.Type = message.NotifyUserStatusMesType
       var notifyUserStatusMes message.NotifyUserStatusMes
       notifyUserStatusMes.UserId = userId
       notifyUserStatusMes.Status = message.UserOnline
       // 将 notifyUserStatusMes 序列化
       data, err := json.Marshal(notifyUserStatusMes)
       if err != nil {
          fmt.Println("json.Marshal err=", err)
          return
       }
       // 将序列化后的 notifyUserStatusMes 赋值给 mes.Data
       mes.Data = string(data)
       // 对 mes 再次序列化，准备发送。
       data, err = json.Marshal(mes)
       if err != nil {
          fmt.Println("json.Marshal err=", err)
          return
       }
       // 创建 Transfer实例
       tf := &utils.Transfer{
          Conn: this.Conn,
       }
       // 发送消息
       err = tf.WritePkg(data)
       if err != nil {
          fmt.Println("NotifyMeOnline err=", err)
          return
       }
    }
    
    // 注册用户
    func (this *UserProcess) ServerProcessRegister(mes *message.Message) (err error) {
       // 1 先从 mes 中取出 mes.Data ，并直接反序列化成 RegisterMes
       var registerMes message.RegisterMes
       err = json.Unmarshal([]byte(mes.Data), &registerMes)
       if err != nil {
          fmt.Println("json.Unmarshal fail err=", err)
          return
       }
       // 1 先声明一个 resMes，用于响应
       var resMes message.Message
       resMes.Type = message.RegisterResMesType
       var registerResMes message.RegisterResMes
       // 需要到 redis 数据库去完成注册。
       // 1 使用 model.MyUserDao 到 redis 去验证。
       err = model.MyUserDao.Register(&registerMes.User)
       if err != nil {
          if err == model.ERROR_USER_EXISTS {
             registerResMes.Code = 505
             registerResMes.Error = model.ERROR_USER_EXISTS.Error()
          } else {
             registerResMes.Code = 506
             registerResMes.Error = "注册发生未知错误"
          }
       } else {
          registerResMes.Code = 200
       }
       data, err := json.Marshal(registerResMes)
       if err != nil {
          fmt.Println("json.Marshal fail", err)
          return
       }
       // 4 将 data 赋值给 resMes
       resMes.Data = string(data)
       // 5 对 resMes 进行序列化，准备发送
       data, err = json.Marshal(resMes)
       if err != nil {
          fmt.Println("json.Marshal fail", err)
          return
       }
       // 6 发送 data 到客户端
       tf := &utils.Transfer{
          Conn: this.Conn,
       }
       err = tf.WritePkg(data)
       return
    }
    
    // 处理登录请求
    func (this *UserProcess) ServerProcessLogin(mes *message.Message) (err error) {
       // 1 先从 mes 中取出 mes.Data ，并直接反序列化成 LoginMes
       var loginMes message.LoginMes
       err = json.Unmarshal([]byte(mes.Data), &loginMes)
       if err != nil {
          fmt.Println("json.Unmarshal fail err=", err)
          return
       }
       // 2 声明一个 resMes
       var resMes message.Message
       resMes.Type = message.LoginResMesType
       // 3 再声明一个 LoginResMes，并完成赋值
       var loginResMes message.LoginResMes
       // 使用 model.MyUserDao 到 redis 去验证
       user, err := model.MyUserDao.Login(loginMes.UserId, loginMes.UserPwd)
       if err != nil {
          if err == model.ERROR_USER_NOTEXISTS {
             loginResMes.Code = 500
             loginResMes.Error = err.Error()
          } else if err == model.ERROR_USER_PWD {
             loginResMes.Code = 403
             loginResMes.Error = err.Error()
          } else {
             loginResMes.Code = 505
             loginResMes.Error = "服务器内部错误"
          }
       } else {
          loginResMes.Code = 200
          // 因为用户登录成功，就把该登录成功的用户放入到 userMgr 中
          // 将登录成功的用户的 userId 赋给 this
          this.UserId = loginMes.UserId
          userMgr.AddOnlineUser(this)
          // 通知其它的在线用户， 我上线了
          this.NotifyOthersOnlineUser(loginMes.UserId)
          // 将当前在线用户的 id 放入到 loginResMes.UsersId
          // 遍历 userMgr.onlineUsers
          for id, _ := range userMgr.onlineUsers {
             loginResMes.UsersId = append(loginResMes.UsersId, id)
          }
          fmt.Println(user, "登录成功")
       }
       // 4 将 loginResMes 序列化
       data, err := json.Marshal(loginResMes)
       if err != nil {
          fmt.Println("json.Marshal fail", err)
          return
       }
       // 5 将 data 赋值给 resMes
       resMes.Data = string(data)
       // 6 对 resMes 进行序列化，准备发送
       data, err = json.Marshal(resMes)
       if err != nil {
          fmt.Println("json.Marshal fail", err)
          return
       }
       // 7 writePkg函数，用于发送 data
       // 因为使用分层模式(mvc), 先创建一个 Transfer 实例，然后发送
       tf := &utils.Transfer{
          Conn: this.Conn,
       }
       err = tf.WritePkg(data)
       return
    }

# 五 main 包 #

**chatroom\\server\\main\\main.go**

package main
    
    import (
       "chatroom/server/model"
       "fmt"
       "net"
       "time"
    )
    
    // 处理和客户端的通讯
    func process(conn net.Conn) {
       // 需要延时关闭 conn
       defer conn.Close()
       // 创建一个总控
       processor := &Processor{
          Conn: conn,
       }
       err := processor.process2()
       if err != nil {
          fmt.Println("客户端和服务器通讯错误=err", err)
          return
       }
    }
    
    // 需要注意一下初始化顺序问题
    // 先 initPool, 再 initUserDao
    func init() {
       // 当服务器启动时，去初始化我们的 redis 的连接池
       initPool("localhost:6379", 16, 0, 300*time.Second)
       initUserDao()
    }
    
    // 完成对 UserDao 的初始化任务
    func initUserDao() {
       // pool 是一个全局的变量
       model.MyUserDao = model.NewUserDao(pool)
    }
    
    func main() {
       // 提示信息
       fmt.Println("服务器在8889端口监听......")
       listen, err := net.Listen("tcp", "0.0.0.0:8889")
       defer listen.Close()
       if err != nil {
          fmt.Println("net.Listen err=", err)
          return
       }
       // 一旦监听成功，就等待客户端来链接服务器
       for {
          fmt.Println("等待客户端来链接服务器......")
          conn, err := listen.Accept()
          if err != nil {
             fmt.Println("listen.Accept err=", err)
          }
          // 一旦链接成功，则启动一个协程和客户端保持通讯
          go process(conn)
       }
    }

**chatroom\\server\\main\\processor.go**

package main
    
    import (
       "chatroom/common/message"
       "chatroom/server/process"
       "chatroom/server/utils"
       "fmt"
       "io"
       "net"
    )
    
    // Processor 的结构体
    type Processor struct {
       Conn net.Conn
    }
    
    // 根据客户端发送消息种类不同，决定调用哪个函数来处理
    func (this *Processor) serverProcessMes(mes *message.Message) (err error) {
       // 接收到客户端发送的群发的消息
       fmt.Println("mes=", mes)
       switch mes.Type {
       case message.LoginMesType:
          // 处理登录
          // 创建一个 UserProcess 实例
          up := &process2.UserProcess{
             Conn: this.Conn,
          }
          err = up.ServerProcessLogin(mes)
       case message.RegisterMesType:
          // 处理注册
          up := &process2.UserProcess{
             Conn: this.Conn,
          }
          err = up.ServerProcessRegister(mes)
       case message.SmsMesType:
          // 创建一个 SmsProcess 实例
          smsProcess := &process2.SmsProcess{}
          // 完成转发群聊消息
          smsProcess.SendGroupMes(mes)
       default:
          fmt.Println("消息类型不存在，无法处理...")
       }
       return
    }
    
    // 消息总控处理
    func (this *Processor) process2() (err error) {
       // 循环处理客户端发送的信息
       for {
          // 创建一个 Transfer 实例
          tf := &utils.Transfer{
             Conn: this.Conn,
          }
          // 读取数据包，封装成一个函数 readPkg(), 返回Message, Err
          mes, err := tf.ReadPkg()
          if err != nil {
             if err == io.EOF {
                fmt.Println("客户端退出，服务器端也退出...")
                return err
             } else {
                fmt.Println("readPkg err=", err)
                return err
             }
          }
          // 用于分发消息
          err = this.serverProcessMes(&mes)
          if err != nil {
             return err
          }
       }
    }

**chatroom\\server\\main\\redis.go**

package main
    
    import (
       "github.com/garyburd/redigo/redis"
       "time"
    )
    
    // 全局的pool
    var pool *redis.Pool
    
    func initPool(address string, maxIdle, maxActive int, idleTimeout time.Duration) {
       pool = &redis.Pool{
          MaxIdle:     maxIdle,     // 最大空闲链接数
          MaxActive:   maxActive,   // 表示和数据库的最大链接数，0 表示没有限制
          IdleTimeout: idleTimeout, // 最大空闲时间
          Dial: func() (redis.Conn, error) { // 初始化连接的代码，连接到哪个 ip 的redis
             return redis.Dial("tcp", address)
          },
       }
    }

[20210523162429999.png]: /images/20210724/d4ad3c7c13ba4234a1b91d2220011415.png