项目架构

简化版本只提供服务端的实现，客户端则使用Linux自带工具nc进行TCP连接

服务器维护一个在线用户map，用户名为key（默认使用用户IP:port），value为User结构体指针

服务器主协程先go一个协程来监听消息广播channel，该channel接收来自客户端发来的消息

服务器主协程上调用listen，accept建立连接后，会go一个协程来运行新建连接的handle函数

在handle函数中，会使用新连接创建新用户，完成用户上线（添加到OnlineMap），然后go一个协程来接收客户端发来的消息，收到客户端发来的消息后会把消息插入到消息广播chaneel中

监听消息广播channel的协程发现channel中有消息后遍历OnlineMap，把消息依次发给每个User用于接收消息的channel，之后User的监听协程会把消息展示给用户

Server.go

提供了Server结构体和一些方法

结构体存储了服务器IP，端口，OnlineMap，用于操作OnlineMap时保证线程安全的读写锁，以及一个接收客户端发来的消息的广播channel

函数NewServer用来创建新Server对象

成员方法ListenMessager用于监听广播channel，如果有消息插入就取出并发送给OnlineMap中所有User

成员方法BroadCast用于将消息插入广播channel

成员方法Handler用于使用连接创建新用户，完成用户上线，并在工作循环协程中接收用户发来的消息并处理，同时在函数末尾有一个空select，作用一是阻塞防止Handler函数结束资源释放，作用而是后续可以扩展方法例如超时强踢

成员方法Start用于启动服务器，会完成listen，accept等操作，创建协程调用以上方法，还defer了close函数，防止资源泄漏

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
}

// 监听广播channel的新消息插入，有消息就广播所有User
func (this *Server) ListenMessager() {
    for {
        msg := <-this.Message

        // 发送消息给所有在线用户
        this.mapLock.Lock()
        for _, client := range this.OnlineMap {
            client.C <- msg
        }
        this.mapLock.Unlock()
    }
}

// 广播消息（将string插入到广播channel）
func (this *Server) BroadCast(user *User, msg string) {
    sendMsg := "[" + user.Addr + "]:" + msg
    // 发送到广播消息channel
    this.Message <- sendMsg
}

// handle函数
func (this *Server) Handler(conn net.Conn) {
    // 新增用户
    user := NewUser(conn, this)

    // 用户上线
    user.Online()

    // 接收客户端发来的消息
    go func() {
        buf := make([]byte, 4096)
        for {
            // 接收用户的消息
            n, err := conn.Read(buf)

            if n == 0 {
                // 用户下线
                user.Offline()
                return
            }

            if err != nil && err != io.EOF {
                fmt.Println("conn Read err:", err)
                return
            }

            msg := string(buf[:n-1])

            // 处理用户发来的消息
            user.DoMessage(msg)
        }
    }()

    // 阻塞
    select {}
}

// 启动server
func (this *Server) Start() {
    // listen
    listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.Ip, this.Port))
    if err != nil {
        fmt.Println("net listen err:", err)
        return
    }
    // close
    defer listener.Close()

    // 启动channel监听
    go this.ListenMessager()

    for {
        // accept
        conn, err := listener.Accept()
        if err != nil {
            fmt.Println("listener accept err:", err)
            continue
        }

        // go一个协程来handle
        go this.Handler(conn)
    }
}

User.go

提供了User结构体和一些方法

User结构体存储了用户名，用户地址，用户的连接（net.Conn类型），用户属于的Server指针以及一个接收广播发来的消息的channel

函数NewUser用于创建User对象

成员方法Online用于上线用户

成员方法Offline用于下线用户

成员方法DoMessage用于处理用户发来的消息并调用Server的广播方法（输入who可以查询在线用户表）

成员方法ListenMessage用于监听广播发来的消息并展示给用户

package main

import "net"

type User struct {
    Name string         // 用户名
    Addr string         // 用户地址
    C    chan string    // 用于接收消息广播的channel
    conn net.Conn       // 用户connection

    server *Server      // 属于的server
}

// 创建User
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,
    }

    // 启动监听广播消息的协程
    go user.ListenMessage()

    return user
}

// 用户上线
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) {
    if msg == "who" {   //  用户查询在线列表
        this.server.mapLock.Lock()
        for _, user := range this.server.OnlineMap {
            onlineMsg := "[" + user.Addr + "]:在线\n"
            this.conn.Write([]byte(onlineMsg))
        }
        this.server.mapLock.Unlock()
    } else {            // 正常广播消息
        this.server.BroadCast(this, msg)
    }
}

// 监听来自channel的广播消息
func (this *User) ListenMessage() {
    for {
        msg := <-this.C
        // 发给用户
        this.conn.Write([]byte(msg + "\n"))
    }
}

main.go

新建了一个Server对象并调用Start()方法

package main

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

效果

编译运行

使用nc命令连接