乐趣区

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

作者：

jiezi

在

即时通信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工程师

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

这个站点使用 Akismet 来减少垃圾评论。了解你的评论数据如何被处理。

更多文章