Golang海量数据通讯系统

实现用户列表

package process


import (
    "encoding/json"
    "fmt"
    "net"
    "os"
    "project_01/client/utils"
    "project_01/common/message"
)


func ShowMenu() {
    fmt.Println("------------恭喜   登录成功-------------")
    fmt.Println("------------1.显示在线用户列表-------------")
    fmt.Println("------------2.发送消息-------------")
    fmt.Println("------------3.信息列表-------------")
    fmt.Println("------------4.退出系统-------------")
    fmt.Println("请选择(1-4):")
    var key int
    fmt.Scanf("%d\n", &key)
    switch key {
    case 1:
        // fmt.Println("显示在线用户列表")
        outputOnlineUser()
    case 2:
        fmt.Println("发送消息")
    case 3:
        fmt.Println("信息列表")
    case 4:
        fmt.Println("你选择推出了系统...")
        os.Exit(0)
    default:
        fmt.Println("你输入的选项不正确")
    }
}


// 和服务器端保持通讯
func ProcessMesServer(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:
            var notifyUserStatusMes message.NotifyUserStatusMes
            json.Unmarshal([]byte(mes.Data), &notifyUserStatusMes)
            updateUserStatus(&notifyUserStatusMes)
        default:
            fmt.Println("服务器返回了未知消息类型")
        }
        // fmt.Println("mes=", mes)
    }
}

package process


import (
    "fmt"
    "project_01/common/message"
)


var onlineUsers map[int]*message.User = make(map[int]*message.User, 10)


// 在客户端显示当前在线用户
func outputOnlineUser() {
    fmt.Println("当前在线用户列表")
    for id, _ := range onlineUsers {
        fmt.Println("用户id:\t", id)
    }
}


func updateUserStatus(notifyUserStatusMes *message.NotifyUserStatusMes) {
    user, ok := onlineUsers[notifyUserStatusMes.UserId]
    if !ok {
        user = &message.User{
            UserId: notifyUserStatusMes.UserId,
        }
    }
    user.UserStatus = notifyUserStatusMes.Status
    onlineUsers[notifyUserStatusMes.UserId] = user


    outputOnlineUser()
}

package process


import (
    "encoding/binary"
    "encoding/json"
    "fmt"
    "net"
    "os"
    "project_01/client/utils"
    "project_01/common/message"
)


type UserProcess struct {
    //字段..


}


func (pc *UserProcess) Register(userId int, userPwd string, userName string) (err error) {
    conn, err := net.Dial("tcp", "localhost:8889")
    if err != nil {
        fmt.Println("net.Dail err=", err)
        return
    }


    defer conn.Close()


    var mes message.Message
    mes.Type = message.RegisterMesType
    var registerMes message.RegisterMes
    registerMes.User.UserId = userId
    registerMes.User.UserPwd = userPwd
    registerMes.User.UserName = userName


    // 将 registerMes 序列化
    data, err := json.Marshal(registerMes.User)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }


    // 把data赋给mes.Data字段
    mes.Data = string(data)


    // 将mes进行序列话
    data, err = json.Marshal(mes)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }


    tf := &utils.Transfer{
        Conn: conn,
    }
    // 发送data给服务器
    err = tf.WritePkg(data)
    if err != nil {
        fmt.Println("注册发送信息错误 err=", err)
    }


    mes, err = tf.ReadPkg() // mes 就是
    if err != nil {
        fmt.Println("readPkg(conn) err=", err)
        return
    }


    // 将mes的Data部分反序列化成 registerMes
    var registerReMes message.RegisterReMes
    err = json.Unmarshal([]byte(mes.Data), &registerReMes)
    if registerReMes.Code == 200 {
        fmt.Println("注册成功！你重新登录一把")
        os.Exit(0)
    } else if registerReMes.Code == 500 {
        fmt.Println(registerReMes.Error)
    }


    return
}


func (pc *UserProcess) Login(userId int, userPwd string) (err error) {
    // 下一步就要开始定协议。。。
    // fmt.Printf("userId=%d\nuserPwd=%s\n", userId, userPwd)
    // return err


    // 1.链接到服务器
    conn, err := net.Dial("tcp", "localhost:8889")
    if err != nil {
        fmt.Println("net.Dail err=", err)
        return
    }


    defer conn.Close()


    var mes message.Message
    mes.Type = message.LoginMesType
    var loginMes message.LoginMes
    loginMes.UserId = userId
    loginMes.UserPwd = userPwd


    // 将loginMes 序列化
    data, err := json.Marshal(loginMes)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }
    // 延时关闭
    defer conn.Close()


    // 把data赋给mes.Data字段
    mes.Data = string(data)


    // 将mes进行序列话
    data, err = json.Marshal(mes)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }
    // data就是我们要发送的信息
    // 先把data的长度发送给服务器
    // 先获取到data长度->转成一个长度的切片
    // conn.Write(len(data))
    pkgLen := uint32(len(data))
    var buf [4]byte
    binary.BigEndian.PutUint32(buf[0:4], pkgLen)
    n, err := conn.Write(buf[:4])
    if n != 4 || err != nil {
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }
    // fmt.Printf("客户端，发送消息的长度=%d\n内容是=%s", len((data)), string(data))


    //  发送消息本身
    _, err = conn.Write(data)
    if err != nil {
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }
    // 还需要处理服务器端返回的消息


    // 创建一个Transfer实例
    tf := &utils.Transfer{
        Conn: conn,
    }
    mes, err = tf.ReadPkg() // mes 就是
    if err != nil {
        fmt.Println("readPkg(conn) err=", err)
        return
    }
    // 将mes的Data部分反序列化成 LoginResMes
    var loginResMes message.LoginResMes
    err = json.Unmarshal([]byte(mes.Data), &loginResMes)
    if loginResMes.Code == 200 {
        // fmt.Println("登陆成功")


        // 显示当前用户列表
        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")
        // 需要在客户端启动一个协程
        // 该协程保持和服务器端的通讯，如果服务器有数据推送给客户端
        go ProcessMesServer(conn)


        // 显示我们登录成功的菜单
        for {
            ShowMenu()
        }


    } else if loginResMes.Code == 500 {
        fmt.Println(loginResMes.Error)


    }
    return
}

package message


// 确定消息类型
const (
    LoginMesType            = "LoginMes"
    LoginResMesType         = "LoginResMes"
    RegisterMesType         = "RegisterMes"
    RegisterReMesType       = "RegisterReMes"
    NotifyUserStatusMesType = "NotifyUserStatusMes"
)


// 用户状态常量
const (
    UserOnline = iota
    UserOffline
    userBusyStatus
)


type Message struct {
    Type string `json:"type"` // 消息的类型
    Data string `hson:"data"` // 消息的内容
}


// 定义两个消息


type LoginMes struct {
    UserId   int    `json:"userId"`
    UserPwd  string `json:"userPwd"`
    UserName string `json:"userName"`
}


type LoginResMes struct {
    Code    int    `json:"code"` // 返回的状态码  500：未注册 200：登陆成功
    UserIds []int  // 保存用户 id 的切片
    Error   string `json:"error"` // 返回错误信息
}


type RegisterMes struct {
    User User `json:"code"`
}


type RegisterReMes struct {
    Code  int    `json:"code"`
    Error string `json:"error"`
}


type NotifyUserStatusMes struct {
    UserId int `json:"userId"` // 用户的id
    Status int `json:"status"` // 用户的状态
}
package message


// 确定消息类型
const (
    LoginMesType            = "LoginMes"
    LoginResMesType         = "LoginResMes"
    RegisterMesType         = "RegisterMes"
    RegisterReMesType       = "RegisterReMes"
    NotifyUserStatusMesType = "NotifyUserStatusMes"
)


// 用户状态常量
const (
    UserOnline = iota
    UserOffline
    userBusyStatus
)


type Message struct {
    Type string `json:"type"` // 消息的类型
    Data string `hson:"data"` // 消息的内容
}


// 定义两个消息


type LoginMes struct {
    UserId   int    `json:"userId"`
    UserPwd  string `json:"userPwd"`
    UserName string `json:"userName"`
}


type LoginResMes struct {
    Code    int    `json:"code"` // 返回的状态码  500：未注册 200：登陆成功
    UserIds []int  // 保存用户 id 的切片
    Error   string `json:"error"` // 返回错误信息
}


type RegisterMes struct {
    User User `json:"code"`
}


type RegisterReMes struct {
    Code  int    `json:"code"`
    Error string `json:"error"`
}


type NotifyUserStatusMes struct {
    UserId int `json:"userId"` // 用户的id
    Status int `json:"status"` // 用户的状态
}

服务端未做修改

实现群聊

package model


import (
    "net"
    "project_01/common/message"
)


// 因为在客户端，我们很多地方会用到curUser，我们将其作为一个全局的
type CurUser struct {
    Conn net.Conn
    message.User
}
package process


import (
    "encoding/json"
    "fmt"
    "net"
    "os"
    "project_01/client/utils"
    "project_01/common/message"
)


func ShowMenu() {
    fmt.Println("------------恭喜   登录成功-------------")
    fmt.Println("------------1.显示在线用户列表-------------")
    fmt.Println("------------2.发送消息-------------")
    fmt.Println("------------3.信息列表-------------")
    fmt.Println("------------4.退出系统-------------")
    fmt.Println("请选择(1-4):")
    var key int
    var content string
    // 因为，我们总会使用到SmsProcsss实例，因此我们将其定义在swtich外部
    smsProcess := &SmsProcess{}
    fmt.Scanf("%d\n", &key)
    switch key {
    case 1:
        // fmt.Println("显示在线用户列表")
        outputOnlineUser()
    case 2:
        fmt.Println("你想对大家说点什么:")
        fmt.Scanf("%s\n", &content)
        smsProcess.sendGorupMes(content)
    case 3:
        fmt.Println("信息列表")
    case 4:
        fmt.Println("你选择推出了系统...")
        os.Exit(0)
    default:
        fmt.Println("你输入的选项不正确")
    }
}


// 和服务器端保持通讯
func ProcessMesServer(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:
            var notifyUserStatusMes message.NotifyUserStatusMes
            json.Unmarshal([]byte(mes.Data), &notifyUserStatusMes)
            updateUserStatus(&notifyUserStatusMes)
        default:
            fmt.Println("服务器返回了未知消息类型")
        }
        // fmt.Println("mes=", mes)
    }
}

package process


import (
    "encoding/json"
    "fmt"
    "project_01/common/message"
    "project_01/server/utils"
)


type SmsProcess struct {
}


// 发送群聊的消息
func (pc *SmsProcess) sendGorupMes(content string) (err error) {
    // 创建一个Mes
    var mes message.Message
    mes.Type = message.SmsMesType


    // 创建一个SmsMes 实例
    var smsMes message.SmsMes
    smsMes.Content = content
    smsMes.UserId = CurUser.UserId
    smsMes.UserStatus = CurUser.UserStatus


    // 序列化smsMes
    data, err := json.Marshal(smsMes)
    if err != nil {
        fmt.Println("SendGroupMes json.Marshal fail = ", err.Error())
        return
    }


    // 将mes发送给服务器
    tf := &utils.Transfer{
        Conn: CurUser.Conn,
    }


    // 发送
    err = tf.WritePkg(data)
    if err != nil {
        fmt.Println("SendGroupMes err=", err.Error())
        return
    }
    return


}

package process


import (
    "encoding/binary"
    "encoding/json"
    "fmt"
    "net"
    "os"
    "project_01/client/utils"
    "project_01/common/message"
)


type UserProcess struct {
    //字段..


}


func (pc *UserProcess) Register(userId int, userPwd string, userName string) (err error) {
    conn, err := net.Dial("tcp", "localhost:8889")
    if err != nil {
        fmt.Println("net.Dail err=", err)
        return
    }


    defer conn.Close()


    var mes message.Message
    mes.Type = message.RegisterMesType
    var registerMes message.RegisterMes
    registerMes.User.UserId = userId
    registerMes.User.UserPwd = userPwd
    registerMes.User.UserName = userName


    // 将 registerMes 序列化
    data, err := json.Marshal(registerMes.User)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }


    // 把data赋给mes.Data字段
    mes.Data = string(data)


    // 将mes进行序列话
    data, err = json.Marshal(mes)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }


    tf := &utils.Transfer{
        Conn: conn,
    }
    // 发送data给服务器
    err = tf.WritePkg(data)
    if err != nil {
        fmt.Println("注册发送信息错误 err=", err)
    }


    mes, err = tf.ReadPkg() // mes 就是
    if err != nil {
        fmt.Println("readPkg(conn) err=", err)
        return
    }


    // 将mes的Data部分反序列化成 registerMes
    var registerReMes message.RegisterReMes
    err = json.Unmarshal([]byte(mes.Data), &registerReMes)
    if registerReMes.Code == 200 {
        fmt.Println("注册成功！你重新登录一把")
        os.Exit(0)
    } else if registerReMes.Code == 500 {
        fmt.Println(registerReMes.Error)
    }


    return
}


func (pc *UserProcess) Login(userId int, userPwd string) (err error) {
    // 下一步就要开始定协议。。。
    // fmt.Printf("userId=%d\nuserPwd=%s\n", userId, userPwd)
    // return err


    // 1.链接到服务器
    conn, err := net.Dial("tcp", "localhost:8889")
    if err != nil {
        fmt.Println("net.Dail err=", err)
        return
    }


    defer conn.Close()


    var mes message.Message
    mes.Type = message.LoginMesType
    var loginMes message.LoginMes
    loginMes.UserId = userId
    loginMes.UserPwd = userPwd


    // 将loginMes 序列化
    data, err := json.Marshal(loginMes)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }
    // 延时关闭
    defer conn.Close()


    // 把data赋给mes.Data字段
    mes.Data = string(data)


    // 将mes进行序列话
    data, err = json.Marshal(mes)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }
    // data就是我们要发送的信息
    // 先把data的长度发送给服务器
    // 先获取到data长度->转成一个长度的切片
    // conn.Write(len(data))
    pkgLen := uint32(len(data))
    var buf [4]byte
    binary.BigEndian.PutUint32(buf[0:4], pkgLen)
    n, err := conn.Write(buf[:4])
    if n != 4 || err != nil {
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }
    // fmt.Printf("客户端，发送消息的长度=%d\n内容是=%s", len((data)), string(data))


    //  发送消息本身
    _, err = conn.Write(data)
    if err != nil {
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }
    // 还需要处理服务器端返回的消息


    // 创建一个Transfer实例
    tf := &utils.Transfer{
        Conn: conn,
    }
    mes, err = tf.ReadPkg() // mes 就是
    if err != nil {
        fmt.Println("readPkg(conn) err=", err)
        return
    }
    // 将mes的Data部分反序列化成 LoginResMes
    var loginResMes message.LoginResMes
    err = json.Unmarshal([]byte(mes.Data), &loginResMes)
    if loginResMes.Code == 200 {
        // 初始化CurUser
        CurUser.Conn = conn
        CurUser.UserId = userId
        CurUser.UserStatus = message.UserOnline


        // fmt.Println("登陆成功")


        // 显示当前用户列表
        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")
        // 需要在客户端启动一个协程
        // 该协程保持和服务器端的通讯，如果服务器有数据推送给客户端
        go ProcessMesServer(conn)


        // 显示我们登录成功的菜单
        for {
            ShowMenu()
        }


    } else if loginResMes.Code == 500 {
        fmt.Println(loginResMes.Error)


    }
    return
}

package message


// 确定消息类型
const (
    LoginMesType            = "LoginMes"
    LoginResMesType         = "LoginResMes"
    RegisterMesType         = "RegisterMes"
    RegisterReMesType       = "RegisterReMes"
    NotifyUserStatusMesType = "NotifyUserStatusMes"
    SmsMesType              = "SmsMes"
)


// 用户状态常量
const (
    UserOnline = iota
    UserOffline
    userBusyStatus
)


type Message struct {
    Type string `json:"type"` // 消息的类型
    Data string `hson:"data"` // 消息的内容
}


// 定义两个消息


type LoginMes struct {
    UserId   int    `json:"userId"`
    UserPwd  string `json:"userPwd"`
    UserName string `json:"userName"`
}


type LoginResMes struct {
    Code    int    `json:"code"` // 返回的状态码  500：未注册 200：登陆成功
    UserIds []int  // 保存用户 id 的切片
    Error   string `json:"error"` // 返回错误信息
}


type RegisterMes struct {
    User User `json:"code"`
}


type RegisterReMes struct {
    Code  int    `json:"code"`
    Error string `json:"error"`
}


type NotifyUserStatusMes struct {
    UserId int `json:"userId"` // 用户的id
    Status int `json:"status"` // 用户的状态
}


// 增加一个SmsMes // 发送的消息
type SmsMes struct {
    Content string `json:"content"`
    User           // 匿名的结构体
}
package message


type User struct {
    UserId     int    `json:"userId"`
    UserPwd    string `json:"userPwd"`
    UserName   string `json:"userName"`
    UserStatus int    `json:"userStantus"`
    Sex        string `json:"sex"`
}

服务端转发消息

完整代码

client/main/main.go

package main


import (
    "fmt"
    "project_01/client/process"
)


var userId int
var userPwd string
var userName string


func main() {
    // 接收用户的选择
    var key int
    // 判断是否还继续显示菜单
    var loop = true
    for loop {


        fmt.Println("------------------欢迎登录多人聊天系统------------------")
        fmt.Println("\t\t\t1.登陆聊天系统")
        fmt.Println("\t\t\t2.注册用户")
        fmt.Println("\t\t\t3.退出系统")
        fmt.Println("------------------------------------------------------")
        fmt.Println("\t\t\t请选择(1-3):")


        fmt.Scanf("%d\n", &key)
        switch key {
        case 1:
            fmt.Println("登陆聊天系统")
            fmt.Println("请输入你的id号")
            fmt.Scanf("%d\n", &userId)
            fmt.Println("请输入你的密码")
            fmt.Scanf("%s\n", &userPwd)
            // loop = false
            // 创建一个UserProcess的实例
            up := &process.UserProcess{}
            up.Login(userId, userPwd)


        case 2:
            fmt.Println("注册用户")
            fmt.Println("请输入用户id:")
            fmt.Scanf("%d\n", &userId)
            fmt.Println("请输入用户密码:")
            fmt.Scanf("%s\n", &userPwd)
            fmt.Println("请输入用户昵称:")
            fmt.Scanf("%s\n", &userName)
            //调用UserProcess，完成注册的请求
            up := &process.UserProcess{}
            up.Register(userId, userPwd, userName)
            // loop = false
        case 3:
            fmt.Println("退出系统")
            // loop = false
        default:
            fmt.Println("输入有误!请重新选择")
        }
    }


}

client/model/curUser.go

package model


import (
    "net"
    "project_01/common/message"
)


// 因为在客户端，我们很多地方会用到curUser，我们将其作为一个全局的
type CurUser struct {
    Conn net.Conn
    message.User
}

client/process/sever.go

package process


import (
    "encoding/json"
    "fmt"
    "net"
    "os"
    "project_01/client/utils"
    "project_01/common/message"
)


func ShowMenu() {
    fmt.Println("------------恭喜   登录成功-------------")
    fmt.Println("------------1.显示在线用户列表-------------")
    fmt.Println("------------2.发送消息-------------")
    fmt.Println("------------3.信息列表-------------")
    fmt.Println("------------4.退出系统-------------")
    fmt.Println("请选择(1-4):")
    var key int
    var content string
    // 因为，我们总会使用到SmsProcsss实例，因此我们将其定义在swtich外部
    smsProcess := &SmsProcess{}
    fmt.Scanf("%d\n", &key)
    switch key {
    case 1:
        // fmt.Println("显示在线用户列表")
        outputOnlineUser()
    case 2:
        fmt.Println("你想对大家说点什么:")
        fmt.Scanf("%s\n", &content)
        smsProcess.sendGorupMes(content)
    case 3:
        fmt.Println("信息列表")
    case 4:
        fmt.Println("你选择推出了系统...")
        os.Exit(0)
    default:
        fmt.Println("你输入的选项不正确")
    }
}


// 和服务器端保持通讯
func ProcessMesServer(Conn net.Conn) {
    // 创建一个Transfer实例，不停的读取服务器发送的消息
    tf := &utils.Transfer{
        Conn: Conn,
    }
    for {
        // wwwfmt.Println("客户端正在等待读取服务器发送的消息")
        mes, err := tf.ReadPkg()
        if err != nil {
            fmt.Println("tf.ReadPkg err=", err)
            return
        }
        // 如果读取到消息，又是下一步处理逻辑
        switch mes.Type {
        case message.NotifyUserStatusMesType:
            var notifyUserStatusMes message.NotifyUserStatusMes
            json.Unmarshal([]byte(mes.Data), &notifyUserStatusMes)
            updateUserStatus(&notifyUserStatusMes)
        case message.SmsMesType:
            OutputGroupMes(&mes)
        default:
            fmt.Println("服务器返回了未知消息类型")
        }
        // fmt.Println("mes=", mes)
    }
}

client/process/smsMgr.go

package process


import (
    "encoding/json"
    "fmt"
    "project_01/common/message"
)


func OutputGroupMes(mes *message.Message) {
    // 反序列化
    var smsMes message.SmsMes
    err := 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/smsProcess.go
package process


import (
    "encoding/json"
    "fmt"
    "project_01/common/message"
)


func OutputGroupMes(mes *message.Message) {
    // 反序列化
    var smsMes message.SmsMes
    err := 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/userMgr.go

package process


import (
    "fmt"
    "project_01/client/model"
    "project_01/common/message"
)


var onlineUsers map[int]*message.User = make(map[int]*message.User, 10)
var CurUser model.CurUser // 我们在用户登陆成功后，完成对CurUser


// 在客户端显示当前在线用户
func outputOnlineUser() {
    fmt.Println("当前在线用户列表")
    for id, _ := range onlineUsers {
        fmt.Println("用户id:\t", id)
    }
}


func updateUserStatus(notifyUserStatusMes *message.NotifyUserStatusMes) {
    user, ok := onlineUsers[notifyUserStatusMes.UserId]
    if !ok {
        user = &message.User{
            UserId: notifyUserStatusMes.UserId,
        }
    }
    user.UserStatus = notifyUserStatusMes.Status
    onlineUsers[notifyUserStatusMes.UserId] = user


    outputOnlineUser()
}

client/process/userProcess.go

package process


import (
    "encoding/binary"
    "encoding/json"
    "fmt"
    "net"
    "os"
    "project_01/client/utils"
    "project_01/common/message"
)


type UserProcess struct {
    //字段..


}


func (pc *UserProcess) Register(userId int, userPwd string, userName string) (err error) {
    conn, err := net.Dial("tcp", "localhost:8889")
    if err != nil {
        fmt.Println("net.Dail err=", err)
        return
    }


    defer conn.Close()


    var mes message.Message
    mes.Type = message.RegisterMesType
    var registerMes message.RegisterMes
    registerMes.User.UserId = userId
    registerMes.User.UserPwd = userPwd
    registerMes.User.UserName = userName


    // 将 registerMes 序列化
    data, err := json.Marshal(registerMes.User)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }


    // 把data赋给mes.Data字段
    mes.Data = string(data)


    // 将mes进行序列话
    data, err = json.Marshal(mes)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }


    tf := &utils.Transfer{
        Conn: conn,
    }
    // 发送data给服务器
    err = tf.WritePkg(data)
    if err != nil {
        fmt.Println("注册发送信息错误 err=", err)
    }


    mes, err = tf.ReadPkg() // mes 就是
    if err != nil {
        fmt.Println("readPkg(conn) err=", err)
        return
    }


    // 将mes的Data部分反序列化成 registerMes
    var registerReMes message.RegisterReMes
    err = json.Unmarshal([]byte(mes.Data), &registerReMes)
    if registerReMes.Code == 200 {
        fmt.Println("注册成功！你重新登录一把")
        os.Exit(0)
    } else if registerReMes.Code == 500 {
        fmt.Println(registerReMes.Error)
    }


    return
}


func (pc *UserProcess) Login(userId int, userPwd string) (err error) {
    // 下一步就要开始定协议。。。
    // fmt.Printf("userId=%d\nuserPwd=%s\n", userId, userPwd)
    // return err


    // 1.链接到服务器
    conn, err := net.Dial("tcp", "localhost:8889")
    if err != nil {
        fmt.Println("net.Dail err=", err)
        return
    }


    defer conn.Close()


    var mes message.Message
    mes.Type = message.LoginMesType
    var loginMes message.LoginMes
    loginMes.UserId = userId
    loginMes.UserPwd = userPwd


    // 将loginMes 序列化
    data, err := json.Marshal(loginMes)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }
    // 延时关闭
    defer conn.Close()


    // 把data赋给mes.Data字段
    mes.Data = string(data)


    // 将mes进行序列话
    data, err = json.Marshal(mes)
    if err != nil {
        fmt.Println("json.Marshal err", err)
        return
    }
    // data就是我们要发送的信息
    // 先把data的长度发送给服务器
    // 先获取到data长度->转成一个长度的切片
    // conn.Write(len(data))
    pkgLen := uint32(len(data))
    var buf [4]byte
    binary.BigEndian.PutUint32(buf[0:4], pkgLen)
    n, err := conn.Write(buf[:4])
    if n != 4 || err != nil {
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }
    // fmt.Printf("客户端，发送消息的长度=%d\n内容是=%s", len((data)), string(data))


    //  发送消息本身
    _, err = conn.Write(data)
    if err != nil {
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }
    // 还需要处理服务器端返回的消息


    // 创建一个Transfer实例
    tf := &utils.Transfer{
        Conn: conn,
    }
    mes, err = tf.ReadPkg() // mes 就是
    if err != nil {
        fmt.Println("readPkg(conn) err=", err)
        return
    }
    // 将mes的Data部分反序列化成 LoginResMes
    var loginResMes message.LoginResMes
    err = json.Unmarshal([]byte(mes.Data), &loginResMes)
    if loginResMes.Code == 200 {
        // 初始化CurUser
        CurUser.Conn = conn
        CurUser.UserId = userId
        CurUser.UserStatus = message.UserOnline


        // fmt.Println("登陆成功")


        // 显示当前用户列表
        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")
        // 需要在客户端启动一个协程
        // 该协程保持和服务器端的通讯，如果服务器有数据推送给客户端
        go ProcessMesServer(conn)


        // 显示我们登录成功的菜单
        for {
            ShowMenu()
        }


    } else if loginResMes.Code == 500 {
        fmt.Println(loginResMes.Error)


    }
    return
}

client/utils/utils.go

package utils


import (
    "encoding/binary"
    "encoding/json"
    "fmt"
    "net"
    "project_01/common/message"
)


// 将方法关联到结构体中
type Transfer struct {
    // 分析有哪些字段
    Conn net.Conn
    Buf  [8096]byte // 传输时，使用缓存
}


func (pc *Transfer) ReadPkg() (mes message.Message, err error) {
    // buf := make([]byte, 8096)
    // fmt.Println("读取服务器发送的数据...")
    // conn.Read 在conn没有关闭的情况下，才会发生阻塞
    // 如果客户端关闭了 conn 则，就不会阻塞
    _, err = pc.Conn.Read(pc.Buf[:4])
    if err != nil {
        // fmt.Println("conn.Read err=", err)
        // err = errors.New("read pkg header error")
        return
    }
    // fmt.Println("独到的buf=", buf)
    pkgLen := binary.BigEndian.Uint32(pc.Buf[0:4])


    //根据 pkgKLen 读取消息内容
    n, err := pc.Conn.Read(pc.Buf[:pkgLen])
    if n != int(pkgLen) || err != nil {
        fmt.Println("conn.Read fail err=", err)
        return
    }


    // 把pkgLen 反序列化成 -> message.Message
    err = json.Unmarshal(pc.Buf[:pkgLen], &mes)
    if err != nil {
        fmt.Println("json.Unmarsha err=", err)
    }


    return
}


func (pc *Transfer) WritePkg(data []byte) (err error) {


    // 先发送一个长度给客户端
    pkgLen := uint32(len(data))
    // var buf [4]byte
    binary.BigEndian.PutUint32(pc.Buf[0:4], pkgLen)
    n, err := pc.Conn.Write(pc.Buf[:4])


    if n != 4 || err != nil {
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }
    // 发送 data 本身
    n, err = pc.Conn.Write(data)


    if n != int(pkgLen) || err != nil {
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }
    return
}

common/message/message.go

package message


// 确定消息类型
const (
    LoginMesType            = "LoginMes"
    LoginResMesType         = "LoginResMes"
    RegisterMesType         = "RegisterMes"
    RegisterReMesType       = "RegisterReMes"
    NotifyUserStatusMesType = "NotifyUserStatusMes"
    SmsMesType              = "SmsMes"
)


// 用户状态常量
const (
    UserOnline = iota
    UserOffline
    userBusyStatus
)


type Message struct {
    Type string `json:"type"` // 消息的类型
    Data string `hson:"data"` // 消息的内容
}


// 定义两个消息


type LoginMes struct {
    UserId   int    `json:"userId"`
    UserPwd  string `json:"userPwd"`
    UserName string `json:"userName"`
}


type LoginResMes struct {
    Code    int    `json:"code"` // 返回的状态码  500：未注册 200：登陆成功
    UserIds []int  // 保存用户 id 的切片
    Error   string `json:"error"` // 返回错误信息
}


type RegisterMes struct {
    User User `json:"code"`
}


type RegisterReMes struct {
    Code  int    `json:"code"`
    Error string `json:"error"`
}


type NotifyUserStatusMes struct {
    UserId int `json:"userId"` // 用户的id
    Status int `json:"status"` // 用户的状态
}


// 增加一个SmsMes // 发送的消息
type SmsMes struct {
    Content string `json:"content"`
    User           // 匿名的结构体
}

common/message/user.go

package message


type User struct {
    UserId     int    `json:"userId"`
    UserPwd    string `json:"userPwd"`
    UserName   string `json:"userName"`
    UserStatus int    `json:"userStantus"`
    Sex        string `json:"sex"`
}

server/main/main.go

package main


import (
    "fmt"
    "net"
    "project_01/server/model"
    "time"
)


func process2(conn net.Conn) {
    // // 延时关闭
    defer conn.Close()


    processor := &Processor{
        Conn: conn,
    }
    err := processor.process3()
    if err != nil {
        fmt.Println("客户端和服务器的通讯协程错误=err", err)
        return
    }
}


// 编写函数，完成对userDao的初始化任务
func initUserDao() {
    model.MyUserDao = model.NewUserDao(pool)
}


func main() {
    // 当服务器启动时，初始化redis的连接池
    initPool("localhost:6379", 8, 0, 300*time.Second)
    initUserDao()
    // 提示信息
    fmt.Println("服务器[新的结构]在8889端口监听...")
    listen, err := net.Listen("tcp", "0.0.0.0:8889")
    // defer listen.Close()
    if err != nil {
        fmt.Println("net.Listen errs=", err)
        return
    }
    for {
        // fmt.Println("等待客户端连接服务器")
        conn, err := listen.Accept()
        if err != nil {
            fmt.Println("Listen.Accept err=", err)
            return
        }
        // 一旦链接成功，则启动一个协程和客户端保持通讯
        go process2(conn)
    }
}

server/main/processor.go

package main


import (
    "fmt"
    "io"
    "net"
    "project_01/common/message"
    "project_01/server/process"
    "project_01/server/utils"
)


// 创建一个Processor结构体
type Processor struct {
    Conn net.Conn
}


// 编写一个ServerProcessMes 函数
// 根据客户端发送消息的种类不同，决定调用那个函数来处理
func (pc *Processor) serverProcessMes(mes *message.Message) (err error) {


    // fmt.Println("mes=", mes)


    switch mes.Type {
    case message.LoginMesType:
        // 处理登录逻辑
        // 创建一个UserProcess实例
        up := &process.UserProcess{
            Conn: pc.Conn,
        }
        err = up.ServerProcessLogin(mes)
    case message.RegisterMesType:
        // 处理注册
        up := &process.UserProcess{
            Conn: pc.Conn,
        }
        err = up.ServerProcessRegister(mes)
    case message.SmsMesType:
        smsProcess := &process.SmsProcess{}
        smsProcess.SendGroupMes(mes)
    default:
        fmt.Println("消息类型不存在，无法处理...")
    }
    return
}


func (pc *Processor) process3() (err error) {
    // 延时关闭
    // defer conn.Close()
    // 读客户端发送的信息
    for {
        // 这里将读取数据包，直接封装成一个函数readPkg(),返回Message，Err
        // 创建一个Transfer，完成读包
        tf := &utils.Transfer{
            Conn: pc.Conn,
        }
        mes, err := tf.ReadPkg()
        if err != nil {
            if err == io.EOF {
                fmt.Println("客户端退出，服务器端也退出..")
                return err
            } else {
                fmt.Println("read Pakage err=", err)
                return err
            }
        }
        // fmt.Println("mes=", mes)


        err = pc.serverProcessMes(&mes)
        if err != nil {
            return err
        }
    }
}

server/main/redis.go

package main


import (
    "time"


    "github.com/garyburd/redigo/redis"
)


var pool *redis.Pool


func initPool(address string, maxIdle, maxActive int, idleTimeout time.Duration) {
    pool = &redis.Pool{
        MaxIdle:     maxIdle,
        MaxActive:   maxActive,
        IdleTimeout: idleTimeout,
        Dial: func() (redis.Conn, error) {
            return redis.Dial("tcp", address)
        },
    }
}

server/model/error.go

package model


import "errors"


var (
    ERROR_USER_NOTEXISTS = errors.New("用户不存在")
    ERROR_USER_EXISTS    = errors.New("用户已经存在")
    ERROR_USER_PWD       = errors.New("密码错误")
)

server/model/userDao.go

package model


import (
    "encoding/json"
    "fmt"
    "project_01/common/message"


    "github.com/garyburd/redigo/redis"
)


// 在服务器启动后，初始化一个userDao实例
var (
    MyUserDao *UserDao
)


type UserDao struct {
    pool *redis.Pool
}


// 使用工厂模式，创建一个UserDao的实例
func NewUserDao(pool *redis.Pool) (userDao *UserDao) {
    userDao = &UserDao{
        pool: pool,
    }
    return
}


func (pc *UserDao) getUserById(conn redis.Conn, id int) (user *message.User, err error) {


    // 通过给定id去redis查询这个用户
    res, err := redis.String(conn.Do("HGet", "user", id))
    if err != nil {
        if err == redis.ErrNil {
            err = ERROR_USER_NOTEXISTS
        }
        return
    }
    user = &message.User{}
    err = json.Unmarshal([]byte(res), user)
    if err != nil {
        fmt.Println("json.Unmarshal err=", err)
    }
    return
}


// Login 完成对用户的验证
// 如果用户的id和pwd都正确，则要返回一个user实例
// 如果用户的id过pwd有错误，则返回对应的错误信息


func (pc *UserDao) Login(userId int, userPwd string) (user *message.User, err error) {
    conn := pc.pool.Get()
    defer conn.Close()
    user, err = pc.getUserById(conn, userId)
    if err != nil {
        return
    }
    if user.UserPwd != userPwd {
        err = ERROR_USER_PWD
        return
    }
    return
}


func (pc *UserDao) Register(user *message.User) (err error) {
    conn := pc.pool.Get()
    defer conn.Close()
    _, err = pc.getUserById(conn, user.UserId)
    if err == nil {
        err = ERROR_USER_EXISTS
        return
    }
    // 说明用户还没出现过
    data, err := json.Marshal(user)
    if err != nil {
        return
    }
    fmt.Print(string(data))
    _, err = conn.Do("Hset", "user", user.UserId, string(data))
    if err != nil {
        fmt.Println("保存注册用户错误 err=", err)
        return
    }
    return
}

server/process/smsProcess.go

package process


import (
    "encoding/json"
    "fmt"
    "net"
    "project_01/common/message"
    "project_01/server/utils"
)


type SmsProcess struct {
}


func (pc *SmsProcess) SendGroupMes(mes *message.Message) {
    // 遍历服务端的onlineUsers map[int]*UserProcess
    // 去除mes的内容
    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.Mashal err=", err)
    }
    for id, up := range userMgr.onlineUsers {
        // 还需要过滤掉自己
        if id == smsMes.UserId {
            continue
        }
        pc.SendMesToEachOnlineUser(string(data), up.Conn)
    }


}


func (pc *SmsProcess) SendMesToEachOnlineUser(data string, conn net.Conn) {


    // 创建Transfer实例，发送data
    tf := &utils.Transfer{
        Conn: conn,
    }
    err := tf.WritePkg([]byte(data))
    if err != nil {
        fmt.Println("转发消息失败 err", err)
    }
}
server/process/userMgr.go
package process


import "fmt"


var (
    userMgr *UserMgr
)


type UserMgr struct {
    onlineUsers map[int]*UserProcess
}


func init() {
    userMgr = &UserMgr{
        onlineUsers: make(map[int]*UserProcess, 1024),
    }
}


// 完成对onlineUsers添加
func (pc *UserMgr) AddOnlineUser(up *UserProcess) {
    pc.onlineUsers[up.UserId] = up


}


// 删除
func (pc *UserMgr) DelOnlineUser(userId int) {
    delete(pc.onlineUsers, userId)
}


// 返回当前所有在线用户
func (pc *UserMgr) GetAllOnlineUser() map[int]*UserProcess {
    return pc.onlineUsers
}


// 根据id返回对应的值
func (pc *UserMgr) GetOnlineUSerById(userId int) (up *UserProcess, err error) {


    // 从map中取出一个值，带检测方式
    up, ok := pc.onlineUsers[userId]
    if !ok {
        err = fmt.Errorf("用户%d 不存在", userId)
        return
    }
    return
}
server/process/userProcess.go
package process


import (
    "encoding/json"
    "fmt"
    "net"
    "project_01/common/message"
    "project_01/server/model"
    "project_01/server/utils"
)


type UserProcess struct {
    Conn   net.Conn
    UserId int
}


// 通知所有在线用户的方法
func (pc *UserProcess) NotifyOthersOnlineUser(userId int) {


    // 遍历
    for id, up := range userMgr.onlineUsers {
        if id == userId {
            continue
        }
        up.NotifyMesOnline(userId)
    }
}


func (pc *UserProcess) NotifyMesOnline(userId int) {


    // 组装
    var mes message.Message
    mes.Type = message.NotifyUserStatusMesType


    var notifyUserStatusMes message.NotifyUserStatusMes
    notifyUserStatusMes.UserId = userId
    notifyUserStatusMes.Status = message.UserOnline


    data, err := json.Marshal(notifyUserStatusMes)
    if err != nil {
        fmt.Println("json.Marshal err=", err)
        return
    }
    mes.Data = string(data)


    //对mes再次序列化
    data, err = json.Marshal(mes)
    if err != nil {
        fmt.Println("json.Marshal err=", err)
        return
    }


    //发送
    tf := &utils.Transfer{
        Conn: pc.Conn,
    }
    err = tf.WritePkg(data)
    if err != nil {
        fmt.Println("NotifyMeOnline err=", err)
        return
    }


}


func (pc *UserProcess) ServerProcessRegister(mes *message.Message) (err error) {
    var registerMes message.RegisterMes
    json.Unmarshal([]byte(mes.Data), &registerMes)
    if err != nil {
        fmt.Println("json.Unmarshal fail err=", err)
        return
    }


    var resMes message.Message
    resMes.Type = message.RegisterReMesType
    var registerReMes message.RegisterReMes


    // 需要到redis数据库去完成注册
    err = model.MyUserDao.Register((*message.User)(&registerMes.User))
    if err != nil {
        if err == model.ERROR_USER_EXISTS {
            registerReMes.Code = 505
            registerReMes.Error = model.ERROR_USER_EXISTS.Error()
        } else {
            registerReMes.Code = 506
            registerReMes.Error = "注册发生未知错误"


        }
    } else {
        registerReMes.Code = 200
    }


    data, err := json.Marshal(registerReMes)
    if err != nil {
        fmt.Println("json.Marshal fail", err)
        return
    }


    // 将 data 赋值给 resMes
    resMes.Data = string(data)
    // 对 resMes 进行序列化，准备发送
    data, err = json.Marshal(resMes)


    if err != nil {
        fmt.Println("json.Marshal fail", err)
        return
    }
    // 发送data ,我们将其封装到writePkg函数中


    // 因为使用了分层模式(mvc)，我们先创建一个Transfer实例，然后读取
    tf := &utils.Transfer{
        Conn: pc.Conn,
    }
    err = tf.WritePkg(data)
    return
}


// 编写一个函数serverProcessLogin函数，专登处理登录请求
func (pc *UserProcess) ServerProcessLogin(mes *message.Message) (err error) {
    // 核心代码。。。
    // 先从mes 中去除mes.Data,并直接反序列化成LoginMes
    var loginMes message.LoginMes
    json.Unmarshal([]byte(mes.Data), &loginMes)
    if err != nil {
        fmt.Println("json.Unmarshal fail err=", err)
        return
    }
    //先声明一个resMes
    var resMes message.Message
    resMes.Type = message.LoginResMesType
    // 在声明一个LoginResMes
    var loginResMes message.LoginResMes


    // 使用model.MyUserDao 到redis去验证
    user, err := model.MyUserDao.Login(loginMes.UserId, loginMes.UserPwd)


    if err != nil {
        loginResMes.Code = 500
        loginResMes.Error = err.Error()
    } else {
        loginResMes.Code = 200
        // 将登陆成功的id -> pc
        pc.UserId = loginMes.UserId
        userMgr.AddOnlineUser(pc)
        pc.NotifyOthersOnlineUser(loginMes.UserId)
        // 将当前在线用户的id，放入到loginResMes。IsersId
        for id, _ := range userMgr.onlineUsers {
            loginResMes.UserIds = append(loginResMes.UserIds, id)
        }


        fmt.Println(user.UserName, "登录成功")
    }
    // // 如果用户id = 123，密码= 123123，认为合法，否则不合法
    // if loginMes.UserId == 123 && loginMes.UserPwd == "123123" {
    //  loginResMes.Code = 200


    // } else {
    //  loginResMes.Code = 500
    //  loginResMes.Error = "该用户不存在，请注册在使用"
    // }


    // 将loginResMes 序列化
    data, err := json.Marshal(loginResMes)
    if err != nil {
        fmt.Println("json.Marshal fail", err)
        return
    }


    // 将 data 赋值给 resMes
    resMes.Data = string(data)
    // 对 resMes 进行序列化，准备发送
    data, err = json.Marshal(resMes)


    if err != nil {
        fmt.Println("json.Marshal fail", err)
        return
    }
    // 发送data ,我们将其封装到writePkg函数中


    // 因为使用了分层模式(mvc)，我们先创建一个Transfer实例，然后读取
    tf := &utils.Transfer{
        Conn: pc.Conn,
    }
    err = tf.WritePkg(data)
    return
}

server/utils/utils.go

package utils


import (
    "encoding/binary"
    "encoding/json"
    "fmt"
    "net"
    "project_01/common/message"
)


// 将方法关联到结构体中
type Transfer struct {
    // 分析有哪些字段
    Conn net.Conn
    Buf  [8096]byte // 传输时，使用缓存
}


func (pc *Transfer) ReadPkg() (mes message.Message, err error) {
    // buf := make([]byte, 8096)
    // fmt.Println("读取客户端发送的数据...")
    // conn.Read 在conn没有关闭的情况下，才会发生阻塞
    // 如果客户端关闭了 conn 则，就不会阻塞
    _, err = pc.Conn.Read(pc.Buf[:4])
    if err != nil {
        // fmt.Println("conn.Read err=", err)
        // err = errors.New("read pkg header error")
        return
    }
    // fmt.Println("独到的buf=", buf)
    pkgLen := binary.BigEndian.Uint32(pc.Buf[0:4])


    //根据 pkgKLen 读取消息内容
    n, err := pc.Conn.Read(pc.Buf[:pkgLen])
    if n != int(pkgLen) || err != nil {
        // fmt.Println("conn.Read fail err=", err)
        return
    }


    // 把pkgLen 反序列化成 -> message.Message
    err = json.Unmarshal(pc.Buf[:pkgLen], &mes)
    if err != nil {
        fmt.Println("json.Unmarsha err=", err)
        return
    }


    return
}


func (pc *Transfer) WritePkg(data []byte) (err error) {


    // 先发送一个长度给客户端
    pkgLen := uint32(len(data))
    // var buf [4]byte
    binary.BigEndian.PutUint32(pc.Buf[0:4], pkgLen)
    n, err := pc.Conn.Write(pc.Buf[:4])
    if n != 4 || err != nil {
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }


    // 发送 data 本身
    n, err = pc.Conn.Write(data)
    if n != int(pkgLen) || err != nil {
        fmt.Println("===================")
        fmt.Println("conn.Write(bytes) fail", err)
        return
    }
    return
}