2.经典项目-海量用户即使通讯系统

1.实现功能-完成注册用户

完成用户注册的步骤(客户端)
1.将User移动到common/message文件夹下
2.在message中新增注册用户的结构体

const (LoginMesType       = "LoginMes"LoginResMesType    = "LoginResMes"RegisterMesType    = "RegisterMes"RegisterResMesType = "RegisterResMes"
)type RegisterMes struct {User User `json:"user"` //类型就是User机构体
}
type RegisterResMes struct {Code  int    `json:"code"`  //返回状态码 400表示该用户已存在 200表示注册成功Error string `json:"error"` //返回错误信息
}

3.在client/process/userProcess.go中添加注册函数

func (this *UserProcess) Register(userId int, userPwd, userName string) (err error) {conn, err := net.Dial("tcp", "localhost:8889")if err != nil {fmt.Println("net.Dial err = ", err)return}defer conn.Close()var mes = message.Message{}mes.Type = message.RegisterMesTyperegisterMes := message.RegisterMes{}registerMes.User.UserId = userIdregisterMes.User.UserPwd = userPwdregisterMes.User.UserName = userNamedata, err := json.Marshal(registerMes)if err != nil {fmt.Println("json.Marshal err = ", err)return}mes.Data = string(data)data, err = json.Marshal(mes)if err != nil {fmt.Println("json.Marshal err = ", err)return}tf := &utils.Transfer{Conn: conn,}err = tf.WritePkg(data)if err != nil {fmt.Println("注册发送信息错误 err = ", err)return}//处理服务器端返回的消息mes, err = tf.ReadPkg() //mes就是 RegisterResMesif err != nil {fmt.Println("utils.ReadPkg(conn) err = ", err)return}var registerResMes message.RegisterResMeserr = json.Unmarshal([]byte(mes.Data), &registerResMes)if registerResMes.Code == 200 {fmt.Println("注册成功，请重新登录")} else {fmt.Println(registerResMes.Error)}return
}

4.在client/main/main.go中修改注册相关代码

		case 2:fmt.Println("注册用户")fmt.Println("请输入用户id:")fmt.Scanf("%d\n", &userId)fmt.Println("请输入用户密码:")fmt.Scanf("%s\n", &userPwd)fmt.Println("请输入用户名称:")fmt.Scanf("%s", &userName)up := &process2.UserProcess{}up.Register(userId, userPwd, userName)

完成用户注册的步骤(服务器端)
1.在model/userDao.go中添加

func (this *UserDao) Register(user *message.User) (err error) {conn := this.pool.Get()defer conn.Close()_, err = this.getUserById(conn, user.UserId)if err == nil {err = ERROR_USER_EXISTSreturn}//账户不存在，则可以正常注册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
}

2.在process/userProcess.go中添加

func (this *UserProcess) ServerProcessRegister(mes *message.Message) (err error) {var registerMes message.RegisterMeserr = json.Unmarshal([]byte(mes.Data), &registerMes)if err != nil {fmt.Println("json.Marshal err = ", err)return}var resMes message.MessageresMes.Type = message.RegisterResMesTypevar registerResMes message.RegisterResMeserr = model.MyUserDao.Register(&registerMes.User)if err != nil {if err == model.ERROR_USER_EXISTS {registerResMes.Code = 505registerResMes.Error = err.Error()} else {registerResMes.Code = 506registerResMes.Error = "注册发生未知错误"}} else {registerResMes.Code = 200fmt.Println("用户注册成功")}data, err := json.Marshal(registerResMes)if err != nil {fmt.Println("json.Marshal err = ", err)return}resMes.Data = string(data)data, err = json.Marshal(resMes)if err != nil {fmt.Println("json.Marshal err = ", err)return}// 发送data，将其封装到writePkg函数//因为使用分层模式（mvc）需要先创建一个Transfer实例，然后读取tf := &utils.Transfer{Conn: this.Conn,}err = tf.WritePkg(data)return
}

3.修改main/processor.go

func (this *Processor) ServerProcessMes(mes *message.Message) (err error) {switch mes.Type {case message.LoginMesType://处理登录//创建UserPorcess实例up := &process2.UserProcess{Conn: this.Conn,}err = up.ServerProcessLogin(mes)case message.RegisterMesType:up := &process2.UserProcess{Conn: this.Conn,}err = up.ServerProcessRegister(mes)default:fmt.Println("消息类型不存在，无法处理...")}return
}

2.实现功能-完成登录时能返回当前在线用户

1.在服务器端维护一个onlineUsers map[int] *UserProcess
2.创建一个新的文件userMgr.go,完成功能对onlineUsers的增删改查
3.在LoginResMess增加一个字段Users []int //保存在线用户id
4.当用户登录后可以显示当前在线用户列表

代码实现
新增server/process/userMgr.go

package processimport "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
}

修改server/process/userProcess.go

type UserProcess struct {Conn net.Conn//增加一个字段，表示该Conn是哪个用户UserId int
}

	} else {loginResMes.Code = 200//将登录成功的用户的userId赋给thisthis.UserId = loginMes.UserId//将登录成功的用户放入userMgr中userMgr.AddOnlineUser(this)//将当前在线用户的id放到loginResMes.UserIds中for id, _ := range userMgr.onlineUsers {loginResMes.UserIds = append(loginResMes.UserIds, id)}fmt.Println(user.UserName, "账户登录成功")

修改message.go

type LoginResMes struct {Code    int    `json:"code"`    //返回状态码 500 表示该用户未注册 200表示登录成功UserIds []int  `json:"userIds"` //增加字段保存userid的切片Error   string `json:"error"`   //返回错误信息
}

修改client/process/userProcess.go

	if loginResMes.Code == 200 {//可以显示当前在线用户列表fmt.Println("当前在线用户列表如下：")for _, v := range loginResMes.UserIds {//不显示自己if v == userId {continue}fmt.Println("用户id:\t", v)}fmt.Print("\n\n")

优化：当一个新的用户上线后，其他已经登录的用户也能获取最新在线用户列表
思路：

1. 当用户A上线，服务器九八A用户的上线信息推给所有在线的用户
2. 客户端也需要维护一个map，map中记录了他的好友（目前就是所有人）map[int]User
3. 客户端和服务器的通讯通道，要依赖serverProcessMes协程

代码实现
在message.go中增加

const (LoginMesType            = "LoginMes"LoginResMesType         = "LoginResMes"RegisterMesType         = "RegisterMes"RegisterResMesType      = "RegisterResMes"NotifyUserStatusMesType = "NotifyUserStatusMes"
)// 定义几个用户状态常量
const (UserOnline = iotaUserOfflineUserBusyStatus
)// 为了配合服务器端推送用户状态变化的消息
type NotifyUserStatusMes struct {UserId int `json:"userId"` //用户idStatus int `json:"status"` //用户状态
}

修改user.go

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

修改server/process/userProcess.go

	} else {loginResMes.Code = 200//将登录成功的用户的userId赋给thisthis.UserId = loginMes.UserId//将登录成功的用户放入userMgr中userMgr.AddOnlineUser(this)//通知其他在线用户this.NotifyOtherOnlineUser(this.UserId)//将当前在线用户的id放到loginResMes.UserIds中for id, _ := range userMgr.onlineUsers {loginResMes.UserIds = append(loginResMes.UserIds, id)}fmt.Println(user.UserName, "账户登录成功")}

在server/process/userProcess.go中增加

// 编写通知所有在线的用户的方法
func (this *UserProcess) NotifyOtherOnlineUser(userId int) {//遍历onlineUsers 然后一个个发送NotifyUserStatusMesfor id, up := range userMgr.onlineUsers {//过滤自己if id == userId {continue}up.NotifyMeOnline(userId)}
}func (this *UserProcess) NotifyMeOnline(userId int) {//组装我们的NotifyUserStatusMesvar mes message.Messagemes.Type = message.NotifyUserStatusMesTypevar notifyUserStatusMes message.NotifyUserStatusMesnotifyUserStatusMes.UserId = userIdnotifyUserStatusMes.Status = message.UserOnline//将notifyUserStatusMes序列化data, err := json.Marshal(notifyUserStatusMes)if err != nil {fmt.Println("json.Marshal err = ", err)return}//将序列化后的notifyUserStatusMes复制给mes.Datames.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}
}

修改client/process/server.go

		case 1:fmt.Println("显示在线用户列表")outpuOnlineUser()

// 和服务器保持通讯
func serverProcessMes(Conn net.Conn) {//创建一个Transfer实例，不停地读取服务器发送的消息tf := &utils.Transfer{Conn: Conn,}for {fmt.Println("客户端正在等待读取服务器发送的消息")mes, err := tf.ReadPkg()if err != nil {fmt.Println("tf.ReadPkg err = ", err)return}//如果读到消息，进入下一步处理逻辑switch mes.Type {case message.NotifyUserStatusMesType: //有人上线//处理//1. 取出NotifyUserStatusMesvar notifyUserStatusMes message.NotifyUserStatusMesjson.Unmarshal([]byte(mes.Data), &notifyUserStatusMes)//2. 把这个用户信息，状态保存到客户map[int]User中updateUserStatus(&notifyUserStatusMes)default:fmt.Println("服务器端返回了未知的消息类型")}//fmt.Printf("mes = %v\n", mes)}
}

新增client/process/userMgt.go

package processimport ("fmt""project/common/message"
)// 客户端维护的map
var onlineUsers map[int]*message.User = make(map[int]*message.User, 10)// 在客户端显示当前在线的用户
func outpuOnlineUser() {//遍历onlilneUsersfmt.Println("当前在线用户列表：")for id, _ := range onlineUsers {fmt.Println("用户id:\t", id)}
}// 编写一个方法，处理返回的NotifyUserStatusMes
func updateUserStatus(notifyUserStatusMes *message.NotifyUserStatusMes) {user, ok := onlineUsers[notifyUserStatusMes.UserId]if !ok {user = &message.User{UserId: notifyUserStatusMes.UserId,}}user.UserStatus = notifyUserStatusMes.StatusonlineUsers[notifyUserStatusMes.UserId] = useroutpuOnlineUser()
}

修改client/process/userProcess.go

	if loginResMes.Code == 200 {//可以显示当前在线用户列表fmt.Println("当前在线用户列表如下：")for _, v := range loginResMes.UserIds {//不显示自己if v == userId {continue}fmt.Println("用户id:\t", v)//完成客户端的onlineUsers初始化user := &message.User{UserId:     v,UserStatus: message.UserOnline,}onlineUsers[v] = user}fmt.Print("\n\n")

3.实现功能-完成登录用户群聊

3.1 完成客户端发送消息

思路
1.新增一个消息结构体smsMes…
2.新增一个model/CurUser
3.在smsProcess.go增加相应的方法SendGroupMes发送一个群聊消息

代码实现
在message中新增

const (LoginMesType            = "LoginMes"LoginResMesType         = "LoginResMes"RegisterMesType         = "RegisterMes"RegisterResMesType      = "RegisterResMes"NotifyUserStatusMesType = "NotifyUserStatusMes"SmsMesType              = "SmsMes"
)// 增加一个SmsMes发送消息
type SmsMes struct {Content string `json:"content"` //消息内容User           //匿名结构体，集继承
}

新增client/model/curUser.go

package modelimport ("net""project/common/message"
)// 在客户端很多地方要用到，需声明为全局
type CurUser struct {Conn net.Connmessage.User
}

在client/process/userMgr.go中新增

var CurUser model.CurUser //在用户登录成功后，完成对CurUser初始化

在client/process/userProcess.go修改

	if loginResMes.Code == 200 {//初始化CurUserCurUser.Conn = connCurUser.UserId = userIdCurUser.UserStatus = message.UserOnline

在client/process/smsProcess.go新增

package processimport ("encoding/json""fmt""project/common/message""project/common/utils"
)type SmsProcess struct {
}// 发送群聊消息
func (this *SmsProcess) SendGroupMes(content string) (err error) {//1.创建一个Mesvar mes message.Messagemes.Type = message.SmsMesType//2.创建一个SmsMes实例var smsMes message.SmsMessmsMes.Content = contentsmsMes.UserId = CurUser.UserIdsmsMes.UserStatus = CurUser.UserStatus//3.序列化smsMesdata, err := json.Marshal(smsMes)if err != nil {fmt.Println("SendGroupMes json.Marshal err = ", err.Error())return}mes.Data = string(data)//4.对mes再次序列化data, err = json.Marshal(mes)if err != nil {fmt.Println("SendGroupMes json.Marshal err = ", err.Error())return}//5.将mes发送给服务器tf := &utils.Transfer{Conn: CurUser.Conn,}//6.发送err = tf.WritePkg(data)if err != nil {fmt.Println("sendGroupMes err = ", err.Error())return}return
}

修改client/process/server.go

		var key intvar content string//用到SmsProcess实例较为频繁，因此定义在外部smsProcess := &SmsProcess{}fmt.Scanf("%d\n", &key)switch key {case 1:fmt.Println("显示在线用户列表")outpuOnlineUser()case 2:fmt.Println("你想对大家说什么:")fmt.Scanf("%s\n", &content)smsProcess.SendGroupMes(content)

3.2 服务器接收群发消息，并发送消息（发送者除外）

思路
1.在服务器端接收到SmsMes消息
2.在server/process/smsProcess.go文件增加群发消息的方法
3.在客户端还要增加去处理服务器转发的群发消息

代码实现
在server/process/smsProcess.go中添加

package processimport ("encoding/json""fmt""net""project/common/message""project/common/utils"
)type SmsProcess struct {
}func (this *SmsProcess) SendGroupMes(mes *message.Message) {var smsMes message.SmsMeserr := 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}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) {tf := &utils.Transfer{Conn: conn,}err := tf.WritePkg(data)if err != nil {fmt.Println("群发消息失败")return}
}

修改server/main/processor.go

func (this *Processor) ServerProcessMes(mes *message.Message) (err error) {switch mes.Type {case message.LoginMesType://处理登录//创建UserPorcess实例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 := &process2.SmsProcess{}smsProcess.SendGroupMes(mes)default:fmt.Println("消息类型不存在，无法处理...")}return
}

新增client/process/smsMgr.go

package processimport ("encoding/json""fmt""project/common/message"
)func outputGroupMes(mes *message.Message) {var smsMes message.SmsMeserr := json.Unmarshal([]byte(mes.Data), &smsMes)if err != nil {fmt.Println("json.Unmarshal err = ", err.Error())return}//显示信息info := fmt.Sprintf("用户id:\t%d 对大家说:\t%s", smsMes.UserId, smsMes.Content)fmt.Println(info)fmt.Println()
}

修改client/process/server.go

		//如果读到消息，进入下一步处理逻辑switch mes.Type {case message.NotifyUserStatusMesType: //有人上线//处理//1. 取出NotifyUserStatusMesvar notifyUserStatusMes message.NotifyUserStatusMesjson.Unmarshal([]byte(mes.Data), &notifyUserStatusMes)//2. 把这个用户信息，状态保存到客户map[int]User中updateUserStatus(&notifyUserStatusMes)case message.SmsMesType: //有人群发消息outputGroupMes(&mes)default:fmt.Println("服务器端返回了未知的消息类型")}