尝试用Go goroutine实现一个简单的聊天服务

news2024/11/30 2:39:49

hello，大家好，我是张张，「架构精进之路」公号作者。

对于聊天服务，想必大家都不会陌生，因为在我们的生活中经常会用到。

我们用 Go 并发来实现一个聊天服务器，这个程序可以让一些用户通过服务器向其它所有用户广播文本消息。

这个程序中有四种 goroutine。
main 和 broadcaster 各自是一个 goroutine 实例，每一个客户端的连接都会有一个handleConn 和 clientWriter 的 goroutine。
broadcaster 是 select 用法的不错的样例，因为它需要处理三种不同类型的消息。

下面我们来演示的 main goroutine 的工作，是 listen 和 accept (网络编程里的概念)从客户端过来的连接。对每一个连接，程序都会建立一个新的 handleConn 的 goroutine。

func main() {
    listener, err := net.Listen("tcp", "localhost:8000")
    if err != nil {
        log.Fatal(err)
    }
    go broadcaster()
    for {
        conn, err := listener.Accept()
        if err != nil {
            log.Print(err)
            continue
        }
        go handleConn(conn)
    }
}

然后是broadcaster的goroutine。

他的内部变量clients会记录当前建立连接的客户端集合，其记录的内容是每一个客户端的消息发出channel的“资格”信息。

type client chan<- string // an outgoing message channel


var (
    entering = make(chan client)
    leaving  = make(chan client)
    messages = make(chan string) // all incoming client messages
)


func broadcaster() {
    clients := make(map[client]bool) // all connected clients
    for {
        select {
        case msg := <-messages:
            // Broadcast incoming message to all
            // clients' outgoing message channels.
            for cli := range clients {
                cli <- msg
            }
        case cli := <-entering:
            clients[cli] = true


        case cli := <-leaving:
            delete(clients, cli)
            close(cli)
        }
    }
}

broadcaster监听来自全局的entering和leaving的channel来获知客户端的到来和离开事件。

当其接收到其中的一个事件时，会更新clients集合，当该事件是离开行为时，它会关闭客户端的消息发送channel。broadcaster也会监听全局的消息channel，所有的客户端都会向这个channel中发送消息。当broadcaster接收到什么消息时，就会将其广播至所有连接到服务端的客户端。

现在让我们看看每一个客户端的goroutine。

handleConn函数会为它的客户端创建一个消息发送channel并通过entering channel来通知客户端的到来。然后它会读取客户端发来的每一行文本，并通过全局的消息channel来将这些文本发送出去，并为每条消息带上发送者的前缀来标明消息身份。当客户端发送完毕后，handleConn会通过leaving这个channel来通知客户端的离开并关闭连接。

func handleConn(conn net.Conn) {
    ch := make(chan string) // outgoing client messages
    go clientWriter(conn, ch)


    who := conn.RemoteAddr().String()
    ch <- "You are " + who
    messages <- who + " has arrived"
    entering <- ch


    input := bufio.NewScanner(conn)
    for input.Scan() {
        messages <- who + ": " + input.Text()
    }
    // NOTE: ignoring potential errors from input.Err()


    leaving <- ch
    messages <- who + " has left"
    conn.Close()
}


func clientWriter(conn net.Conn, ch <-chan string) {
    for msg := range ch {
        fmt.Fprintln(conn, msg) // NOTE: ignoring network errors
    }
}

另外，handleConn为每一个客户端创建了一个clientWriter的goroutine，用来接收向客户端发送消息的channel中的广播消息，并将它们写入到客户端的网络连接。客户端的读取循环会在broadcaster接收到leaving通知并关闭了channel后终止。

下面演示的是当服务器有两个活动的客户端连接，并且在两个窗口中运行的情况，使用netcat来聊天：

$ go build gopl.io/ch8/chat
$ go build gopl.io/ch8/netcat3
$ ./chat &
$ ./netcat3
You are 127.0.0.1:64208               $ ./netcat3
127.0.0.1:64211 has arrived           You are 127.0.0.1:64211
Hi!
127.0.0.1:64208: Hi!                  127.0.0.1:64208: Hi!
                                      Hi yourself.
127.0.0.1:64211: Hi yourself.         127.0.0.1:64211: Hi yourself.
^C
                                      127.0.0.1:64208 has left
$ ./netcat3
You are 127.0.0.1:64216               127.0.0.1:64216 has arrived
                                      Welcome.
127.0.0.1:64211: Welcome.             127.0.0.1:64211: Welcome.
                                      ^C
127.0.0.1:64211 has left”

当与n个客户端保持聊天session时，这个程序会有2n+2个并发的goroutine，然而这个程序却并不需要显式的锁。clients这个map被限制在了一个独立的goroutine中，broadcaster，所以它不能被并发地访问。