关于golang:Go笔记03即时通信Demo

43次阅读

共计 14490 个字符，预计需要花费 37 分钟才能阅读完成。

即时通信 Demo

V0.1 根底 Server 构建

实现新用户连贯胜利后，在 server 端揭示的性能

server.go

package main
import (
    "fmt"
    "net"
)
type Server struct {
    Ip string
    Port int
}
// 创立一个 server 的接口
func NewServer(ip string,port int) *Server {
    server := &Server{
        Ip: ip,
        Port: port,
    }
    return server
}
// 启动服务器的接口
// 定义对象 (类) 的办法：func (对象类型参数)办法名(参数列表)(返回值列表){ }
// 如果想通过办法批改对象，那么倡议传递对象的地址（构造体是值传递，通过构造体的指针批改构造体(地址传递)）（也能够通过返回值批改对象）//func (obj *MyInt) add() {}  // 对象调用时，会主动将对象的地址传给 obj
func (this *Server) Start() {
    //socket listen
    listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d",this.Ip,this.Port))
    if err != nil {fmt.Println("net.listen tcp err:", err)
    }
    //close listen socket
    defer listener.Close()
    for {
        //accept
        conn, err := listener.Accept()
        if err != nil {fmt.Println("listen accept err:", err)
            continue
        }
        //do handler
        go this.Handler(conn)    
    }
}
// 链接胜利，执行业务
func (this *Server) Handler(conn net.Conn) {fmt.Println("Connection Success...")
}

main.go

package main
func main()  {server := NewServer("127.0.0.1", 8888)
    server.Start()}

编译

go build -o server main.go server.go

开启 server

./server

测试连贯

nc 127.0.0.1 8888

V0.2 用户上线及播送

实现新用户连贯胜利后，向整体在线用户播送的性能

server.go

package main
import (
    "fmt"
    "net"
    "sync"
)
type Server struct {
    Ip   string
    Port int
    // 在线用户列表
    OnlineMap map[string]*User
    mapLock   sync.RWMutex
    // 音讯播送 channel
    Message chan string
}
// 创立一个 server 的接口
func NewServer(ip string, port int) *Server {
    server := &Server{
        Ip:        ip,
        Port:      port,
        OnlineMap: make(map[string]*User),
        Message:   make(chan string),
    }
    return server
}
// 坚挺 Message 播送，有音讯则发送给全副在线 User (在 start 办法中加载)
func (this *Server) ListenMessager() {
    for {
        msg := <-this.Message
        this.mapLock.Lock()
        for _, user := range this.OnlineMap {user.C <- msg}
        this.mapLock.Unlock()}
}
// 音讯播送
func (this *Server) BroadCast(user *User, msg string) {sendMsg := "[" + user.Addr + "]" + user.Name + ":" + msg
    this.Message <- sendMsg
}

// 链接胜利，执行业务
func (this *Server) Handler(conn net.Conn) {user := NewUser(conn)
    // 用户上线，将用户退出 OnlineMap 中
    this.mapLock.Lock()
    this.OnlineMap[user.Name] = user
    this.mapLock.Unlock()
    // 播送用户上线音讯
    this.BroadCast(user, "上线了")
    // 阻塞以后 Handler(避免以后及子 goroutine 死亡)
    select {}}
// 启动服务器的接口
func (this *Server) Start() {
    //socket listen
    listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.Ip, this.Port))
    if err != nil {fmt.Println("net.listen tcp err:", err)
    }
    //close listen socket
    defer listener.Close()
    // 启动监听 Message 的 goroutine
    go this.ListenMessager()
    for {
        //accept
        conn, err := listener.Accept()
        if err != nil {fmt.Println("listen accept err:", err)
            continue
        }
        //do handler
        go this.Handler(conn)
    }
}

user.go

package main
import "net"
type User struct {
    Name string
    Addr string
    C    chan string
    conn net.Conn
}
// 创立用户
func NewUser(conn net.Conn) *User {userAddr := conn.RemoteAddr().String()
    user := &User{
        Name: userAddr,
        Addr: userAddr,
        C:    make(chan string),
        conn: conn,
    }
    // 启动监听以后 user channel 的 goroutine
    go user.ListenMessage()
    return user
}
// 监听以后 User channel，有新音讯则发送到客户端
func (this *User) ListenMessage() {
    for {
        msg := <-this.C
        this.conn.Write([]byte(msg + "\n"))
    }
}

main.go

package main
func main()  {server := NewServer("127.0.0.1", 8888)
    server.Start()}

编译运行

go build -o server main.go server.go user.go

连贯测试

nc 127.0.0.1 8888

V0.3 用户音讯播送

server.go

func (this *Server) Handler(conn net.Conn) {user := NewUser(conn)
    // 用户上线，将用户退出 OnlineMap 中
    this.mapLock.Lock()
    this.OnlineMap[user.Name] = user
    this.mapLock.Unlock()
    // 播送用户上线音讯
    this.BroadCast(user, "上线了")
    // 接管用户音讯
    go func() {buf := make([]byte, 4096)
        for {
            // 将 conn 中数据读取到 buf 中
            n, err := conn.Read(buf)
            if n == 0 {this.BroadCast(user, "下线")
                return
            }
            if err != nil && err != io.EOF {fmt.Println("Conn Read err:", err)
                return
            }
            // 提取用户的音讯（去除 \n）msg := string(buf[:n-1])
            // 将失去的音讯进行播送
            this.BroadCast(user, msg)
        }
    }()
    // 阻塞以后 Handler(避免以后及子 goroutine 死亡)
    select {}}

查看端口占用：

lsof -i:8888

V0.4 用户业务封装

Server.go

package main

import (
    "fmt"
    "io"
    "net"
    "sync"
)

type Server struct {
    Ip   string
    Port int
    // 在线用户列表
    OnlineMap map[string]*User
    mapLock   sync.RWMutex
    // 音讯播送 channel
    Message chan string
}

// 创立一个 server 的接口
func NewServer(ip string, port int) *Server {
    server := &Server{
        Ip:        ip,
        Port:      port,
        OnlineMap: make(map[string]*User),
        Message:   make(chan string),
    }
    return server
}

// 坚挺 Message 播送，有音讯则发送给全副在线 User (在 start 办法中加载)
func (this *Server) ListenMessager() {
    for {
        msg := <-this.Message
        this.mapLock.Lock()
        for _, user := range this.OnlineMap {user.C <- msg}
        this.mapLock.Unlock()}
}

// 音讯播送
func (this *Server) BroadCast(user *User, msg string) {sendMsg := "[" + user.Addr + "]" + user.Name + ":" + msg
    this.Message <- sendMsg
}

// 链接胜利，执行业务
func (this *Server) Handler(conn net.Conn) {user := NewUser(conn, this)
    // 用户上线，将用户退出 OnlineMap 中
    user.Online()
    // 接管用户音讯
    go func() {buf := make([]byte, 4096)
        for {
            // 将 conn 中数据读取到 buf 中
            n, err := conn.Read(buf)
            if n == 0 {user.Offline()
                return
            }
            if err != nil && err != io.EOF {fmt.Println("Conn Read err:", err)
                return
            }
            // 提取用户的音讯（去除 \n）msg := string(buf[:n-1])
            // 将失去的音讯进行播送
            user.DoMessage(msg)
        }
    }()
    // 阻塞以后 Handler(避免以后及子 goroutine 死亡)
    select {}}

// 启动服务器的接口
func (this *Server) Start() {
    //socket listen
    listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.Ip, this.Port))
    if err != nil {fmt.Println("net.listen tcp err:", err)
    }
    //close listen socket
    defer listener.Close()
    // 启动监听 Message 的 goroutine
    go this.ListenMessager()
    for {
        //accept
        conn, err := listener.Accept()
        if err != nil {fmt.Println("listen accept err:", err)
            continue
        }
        //do handler
        go this.Handler(conn)
    }
}

user.go

package main

import "net"

type User struct {
    Name   string
    Addr   string
    C      chan string
    conn   net.Conn
    server *Server
}

// 创立用户
func NewUser(conn net.Conn, server *Server) *User {userAddr := conn.RemoteAddr().String()
    user := &User{
        Name:   userAddr,
        Addr:   userAddr,
        C:      make(chan string),
        conn:   conn,
        server: server,
    }
    // 启动监听以后 user channel 的 goroutine
    go user.ListenMessage()
    return user
}

// 监听以后 User channel，有新音讯则发送到客户端
func (this *User) ListenMessage() {
    for {
        msg := <-this.C
        this.conn.Write([]byte(msg + "\n"))
    }
}

// 用户上线
func (this *User) Online() {this.server.mapLock.Lock()
    this.server.OnlineMap[this.Name] = this
    this.server.mapLock.Unlock()
    this.server.BroadCast(this, "上线了")
}

// 用户下线
func (this *User) Offline() {this.server.mapLock.Lock()
    delete(this.server.OnlineMap, this.Name)
    this.server.mapLock.Unlock()
    this.server.BroadCast(this, "下线了")
}

// 用户音讯
func (this *User) DoMessage(msg string) {this.server.BroadCast(this, msg)
}

V0.5 在线用户查问

user.go

// 用户音讯
func (this *User) DoMessage(msg string) {
    if msg == "who" {this.server.OnlineMap.Lock()
        for _, user := range this.server.OnlineMap {olMsg := "[" + user.Addr + "]" + user.Name + ": 在线...\n"
            this.conn.Write([]byte(olMsg))
        }
        this.server.OnlineMap.Unlock()} else {this.server.BroadCast(this, msg)
    }
}

V0.6 批改用户名

user.go

// 向客户端输入音讯
func (this *User) sendMsg(msg string) {this.conn.Write([]byte(msg))
}
// 用户音讯
func (this *User) DoMessage(msg string) {
    if msg == "who" {this.server.mapLock.Lock()
        for _, user := range this.server.OnlineMap {olMsg := "[" + user.Addr + "]" + user.Name + ": 在线...\n"
            this.sendMsg(olMsg)
        }
        this.server.mapLock.Unlock()} else if len(msg) > 7 && msg[:7] == "rename|" {newName := strings.Split(msg, "|")[1]
        _, ok := this.server.OnlineMap[newName]
        if ok {this.sendMsg("以后用户名已被应用 \n")
        } else {this.server.mapLock.Lock()
            delete(this.server.OnlineMap, this.Name)
            this.server.OnlineMap[newName] = this
            this.server.mapLock.Unlock()
            this.Name = newName
            this.sendMsg("您曾经更名为：" + newName + "\n")
        }
    } else {this.server.BroadCast(this, msg)
    }
}

V0.7 超时强踢

server.go

// 链接胜利，执行业务
func (this *Server) Handler(conn net.Conn) {user := NewUser(conn, this)
    // 用户上线，将用户退出 OnlineMap 中
    user.Online()
    // 监听用户是否沉闷的 channel
    isLive := make(chan bool)
    // 接管用户音讯
    go func() {buf := make([]byte, 4096)
        for {
            // 将 conn 中数据读取到 buf 中
            n, err := conn.Read(buf)
            if n == 0 {user.Offline()
                return
            }
            if err != nil && err != io.EOF {fmt.Println("Conn Read err:", err)
                return
            }
            // 提取用户的音讯（去除 \n）msg := string(buf[:n-1])
            // 将失去的音讯进行播送
            user.DoMessage(msg)
            // 用户任意音讯
            isLive <- true
        }
    }()
    // 阻塞以后 Handler(避免以后及子 goroutine 死亡)
    for {
        select {
        case <-isLive:
            // 以后用户是沉闷的，激活 select 重置定时器
        case <-time.After(time.Second * 10):
            // 倒计时 10 秒强制退出
            user.sendMsg("你被踢了")
            close(user.C)
            conn.Close()
            return
        }
    }
}

V0.8 私聊性能

// 用户音讯
func (this *User) DoMessage(msg string) {...else if len(msg) > 4 && msg[:3] == "to|" {remoteName := strings.Split(msg, "|")[1]
        if remoteName == "" {this.sendMsg("音讯格局不正确 \n")
            return
        }
        // 获取对方 User 对象
        remoteUser, ok := this.server.OnlineMap[remoteName]
        if !ok {this.sendMsg("用户不存在 \n")
            return
        }
        // 音讯内容发送给指定 User
        content := strings.Split(msg, "|")[2]
        if content == "" {this.sendMsg("音讯不能为空 \n")
            return
        }
        remoteUser.sendMsg(this.Name + "对您说：" + content + "\n")
    }...
}

V0.9 客户端实现 client.go

建设连贯
package main
import (
    "fmt"
    "net"
)
type Client struct {
    ServerIp   string
    ServerPort int
    Name       string
    conn       net.Conn
}
func NewClient(serverIp string, serverPort int) *Client {
    client := &Client{
        ServerIp:   serverIp,
        ServerPort: serverPort,
    }
    conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", serverIp, serverPort))
    if err != nil {fmt.Println("net.Dial err:", err)
        return nil
    }
    client.conn = conn
    return client
}
func main() {client := NewClient("127.0.0.1", 8888)
    if client == nil {fmt.Println(">>>>> 连贯服务器失败")
        return
    }
    fmt.Println(">>>>> 连贯服务器胜利")
    // 阻塞服务 
    select {}}

测试

go build -o client client.go
./client
解析命令行
package main

import (
    "flag"
    "fmt"
    "net"
)

type Client struct {
    ServerIp   string
    ServerPort int
    Name       string
    conn       net.Conn
}

var serverIp string
var serverPort int

// ./client -ip 127.0.0.1 -port 8888
func init() {flag.StringVar(&serverIp, "ip", "127.0.0.1", "设置服务器 IP")
    flag.IntVar(&serverPort, "port", 8888, "设置服务器端口")
}

func NewClient(serverIp string, serverPort int) *Client {
    client := &Client{
        ServerIp:   serverIp,
        ServerPort: serverPort,
    }
    conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", serverIp, serverPort))
    if err != nil {fmt.Println("net.Dial err:", err)
        return nil
    }
    client.conn = conn
    return client
}

func main() {flag.Parse()
    client := NewClient(serverIp, serverPort)
    if client == nil {fmt.Println(">>>>> 连贯服务器失败")
        return
    }
    fmt.Println(">>>>> 连贯服务器胜利")
    select {}}

测试

go build -o client client.go
./client -ip 127.0.0.1 -port 8888
菜单显示
package main

import (
    "flag"
    "fmt"
    "net"
)

type Client struct {
    ServerIp   string
    ServerPort int
    Name       string
    conn       net.Conn
    flag       int // 以后 client 的模式
}

func (client *Client) Run() {
    // 除非输出 0 否则始终循环
    for client.flag != 0 {
        // 除非输出 0 -3，否则始终循环
        for client.menu() != true {}
        // 依据 flag 执行不同业务
        switch client.flag {
        case 1:
            fmt.Println("抉择公聊...")
        case 2:
            fmt.Println("抉择私聊...")
        case 3:
            fmt.Println("抉择改名...")
        }
    }
}
func (client *Client) menu() bool {
    var flag int // 接管用户输出
    fmt.Println("1. 公聊")
    fmt.Println("2. 私聊")
    fmt.Println("3. 改名")
    fmt.Println("0. 退出")
    fmt.Scanln(&flag)
    if flag >= 0 && flag <= 3 {
        client.flag = flag
        return true
    } else {fmt.Println("输出不非法")
        return false
    }
}

var serverIp string
var serverPort int

// ./client -ip 127.0.0.1 -port 8888
func init() {flag.StringVar(&serverIp, "ip", "127.0.0.1", "设置服务器 IP")
    flag.IntVar(&serverPort, "port", 8888, "设置服务器端口")
}

func NewClient(serverIp string, serverPort int) *Client {
    client := &Client{
        ServerIp:   serverIp,
        ServerPort: serverPort,
        flag:       999,
    }
    conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", serverIp, serverPort))
    if err != nil {fmt.Println("net.Dial err:", err)
        return nil
    }
    client.conn = conn
    return client
}

func main() {flag.Parse()
    client := NewClient(serverIp, serverPort)
    if client == nil {fmt.Println(">>>>> 连贯服务器失败")
        return
    }
    fmt.Println(">>>>> 连贯服务器胜利")
    // 启动客户端业务
    client.Run()}

测试

go build -o client client.go
./client -ip 127.0.0.1 -port 8888
更新用户名
package main

import (
    "flag"
    "fmt"
    "io"
    "net"
    "os"
)

type Client struct {
    ServerIp   string
    ServerPort int
    Name       string
    conn       net.Conn
    flag       int // 以后 client 的模式
}

func (client *Client) Run() {
    // 除非输出 0 否则始终循环
    for client.flag != 0 {
        // 除非输出 0 -3，否则始终循环
        for client.menu() != true {}
        // 依据 flag 执行不同业务
        switch client.flag {
        case 1:
            fmt.Println("抉择公聊...")
        case 2:
            fmt.Println("抉择私聊...")
        case 3:
            client.UpdateName()}
    }
}
func (client *Client) menu() bool {
    var flag int // 接管用户输出
    fmt.Println("1. 公聊")
    fmt.Println("2. 私聊")
    fmt.Println("3. 改名")
    fmt.Println("0. 退出")
    fmt.Scanln(&flag)
    if flag >= 0 && flag <= 3 {
        client.flag = flag
        return true
    } else {fmt.Println("输出不非法")
        return false
    }
}

var serverIp string
var serverPort int

// ./client -ip 127.0.0.1 -port 8888
func init() {flag.StringVar(&serverIp, "ip", "127.0.0.1", "设置服务器 IP")
    flag.IntVar(&serverPort, "port", 8888, "设置服务器端口")
}

func NewClient(serverIp string, serverPort int) *Client {
    client := &Client{
        ServerIp:   serverIp,
        ServerPort: serverPort,
        flag:       999,
    }
    conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", serverIp, serverPort))
    if err != nil {fmt.Println("net.Dial err:", err)
        return nil
    }
    client.conn = conn
    return client
}
func (client *Client) UpdateName() bool {fmt.Println(">>>>> 请输出用户名：")
    fmt.Scanln(&client.Name)
    sendMsg := "rename|" + client.Name + "\n"
    _, err := client.conn.Write([]byte(sendMsg))
    if err != nil {fmt.Println("conn.Write err:", err)
        return false
    }
    return true
}

// 解决 server 回应的音讯，间接显示到规范输入即可
func (client *Client) DealResponse() {
    // 一旦 client 有数据，就间接 copy 到 stdout 规范输入上，永恒阻塞监听
    io.Copy(os.Stdout, client.conn)
}
func main() {flag.Parse()
    client := NewClient(serverIp, serverPort)
    if client == nil {fmt.Println(">>>>> 连贯服务器失败")
        return
    }
    fmt.Println(">>>>> 连贯服务器胜利")
    // 独自开启一个 goroutine 解决 server 的回执音讯
    go client.DealResponse()
    // 启动客户端业务
    client.Run()}
公聊模式
func (client *Client) PublicChat() {
    var chatMsg string
    fmt.Println(">>>>> 请输出聊天内容，exit 退出")
    fmt.Scanln(&chatMsg)
    for chatMsg != "exit" {if len(chatMsg) != 0 {
            sendMsg := chatMsg + "\n"
            _, err := client.conn.Write([]byte(sendMsg))
            if err != nil {fmt.Println("conn.Write err:", err)
                break
            }
            chatMsg = ""fmt.Println(">>>>> 请输出聊天内容，exit 退出 ")
            fmt.Scanln(&chatMsg)
        }
    }
}
私聊模式
package main

import (
    "flag"
    "fmt"
    "io"
    "net"
    "os"
)

type Client struct {
    ServerIp   string
    ServerPort int
    Name       string
    conn       net.Conn
    flag       int // 以后 client 的模式
}

func (client *Client) Run() {
    // 除非输出 0 否则始终循环
    for client.flag != 0 {
        // 除非输出 0 -3，否则始终循环
        for client.menu() != true {}
        // 依据 flag 执行不同业务
        switch client.flag {
        case 1:
            client.PublicChat()
        case 2:
            client.PrivateChat()
        case 3:
            client.UpdateName()}
    }
}
func (client *Client) menu() bool {
    var flag int // 接管用户输出
    fmt.Println("1. 公聊")
    fmt.Println("2. 私聊")
    fmt.Println("3. 改名")
    fmt.Println("0. 退出")
    fmt.Scanln(&flag)
    if flag >= 0 && flag <= 3 {
        client.flag = flag
        return true
    } else {fmt.Println("输出不非法")
        return false
    }
}

var serverIp string
var serverPort int

// ./client -ip 127.0.0.1 -port 8888
func init() {flag.StringVar(&serverIp, "ip", "127.0.0.1", "设置服务器 IP")
    flag.IntVar(&serverPort, "port", 8888, "设置服务器端口")
}

func NewClient(serverIp string, serverPort int) *Client {
    client := &Client{
        ServerIp:   serverIp,
        ServerPort: serverPort,
        flag:       999,
    }
    conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", serverIp, serverPort))
    if err != nil {fmt.Println("net.Dial err:", err)
        return nil
    }
    client.conn = conn
    return client
}
func (client *Client) PublicChat() {
    var chatMsg string
    fmt.Println(">>>>> 请输出聊天内容，exit 退出")
    fmt.Scanln(&chatMsg)
    for chatMsg != "exit" {if len(chatMsg) != 0 {
            sendMsg := chatMsg + "\n"
            _, err := client.conn.Write([]byte(sendMsg))
            if err != nil {fmt.Println("conn.Write err:", err)
                break
            }
            chatMsg = ""fmt.Println(">>>>> 请输出聊天内容，exit 退出 ")
            fmt.Scanln(&chatMsg)
        }
    }
}
func (client *Client) SelectUsers() {
    sendMsg := "who\n"
    _, err := client.conn.Write([]byte(sendMsg))
    if err != nil {fmt.Println("conn.Write err:", err)
        return
    }
}

func (client *Client) PrivateChat() {
    var remoteName string
    var chatMsg string
    client.SelectUsers()
    fmt.Println(">>>>> 请输出用户名，exit 退出")
    fmt.Scanln(&remoteName)
    for remoteName != "exit" {fmt.Println(">>>>> 请输出聊天内容，exit 退出")
        fmt.Scanln(&chatMsg)
        for chatMsg != "exit" {if len(chatMsg) != 0 {
                sendMsg := "to|" + remoteName + "|" + chatMsg + "\n\n"
                _, err := client.conn.Write([]byte(sendMsg))
                if err != nil {fmt.Println("conn.Write err:", err)
                    break
                }
                chatMsg = ""fmt.Println(">>>>> 请输出聊天内容，exit 退出 ")
                fmt.Scanln(&chatMsg)
            }
        }
        client.SelectUsers()
        fmt.Println(">>>>> 请输出用户名，exit 退出")
        fmt.Scanln(&remoteName)
    }
}

func (client *Client) UpdateName() bool {fmt.Println(">>>>> 请输出用户名：")
    fmt.Scanln(&client.Name)
    sendMsg := "rename|" + client.Name + "\n"
    _, err := client.conn.Write([]byte(sendMsg))
    if err != nil {fmt.Println("conn.Write err:", err)
        return false
    }
    return true
}

// 解决 server 回应的音讯，间接显示到规范输入即可
func (client *Client) DealResponse() {
    // 一旦 client 有数据，就间接 copy 到 stdout 规范输入上，永恒阻塞监听
    io.Copy(os.Stdout, client.conn)
}
func main() {flag.Parse()
    client := NewClient(serverIp, serverPort)
    if client == nil {fmt.Println(">>>>> 连贯服务器失败")
        return
    }
    fmt.Println(">>>>> 连贯服务器胜利")
    // 独自开启一个 goroutine 解决 server 的回执音讯
    go client.DealResponse()
    // 启动客户端业务
    client.Run()}

参考资料

8 小时转职 Golang 工程师

正文完
golang
发表至： golang
2021-12-05
 0
关于golang:docker-搭建-Elasticsearch-和-Kibana
关于golang:是的没错我就是抄的一个像-Laravel-那样好用的-go-语言的-SQL-查询构造器
关于golang:golang-中接口-指针类型和值类型方法区别
关于golang:Golang内部构件第1部分主要概念和项目结构