培训-golang-高级编程并发聊天室
并发聊天室#
并发编程和网络编程是现今行业开发中常用的技术。Go 语言强大的语法设定使得并发和网络编程都变的简洁而高效。
下面我们利用前面学到的知识，使用并发和网络实现一个简单的网络在线聊天室。体会下这两种技术的实际应用。在整个聊天室的项目中，充分利用了 go 程并发，处理不同任务。
整个聊天室程序可简单划分为如下模块，都分别使用 go 程来实现：
主 go** 程（服务器）：**
负责监听、接收用户（客户端）连接请求，建立通信关系。同时启动相应的 go 程处理任务。
处理连接用户数据 go** 程：HandleConnect **
负责新上线用户的存储，用户消息读取、发送，用户改名、下线处理及超时处理。
为了提高并发效率，同时给一个用户维护多个 go 程来并行处理上述任务。
用户消息广播 go** 程：Manager**
负责在线用户遍历，用户消息广播发送。需要与 HandleConnect go 程及用户子 go 程协作完成。
go** 程间应用数据及通信：**
map：存储所有登录聊天室的用户信息， key：用户的 ip+port。Value：Client 结构体。
Client 结构体：包含成员：用户名 Name，网络地址 Addr（ip+port），发送消息的通道 C（channel）
通道 message：协调并发 go 程间消息的传递。
广播用户上线#
首先，服务器启动，等待用户建立通信连接。当有用户连接上来，将其存储到 map 中，这样就维护了一个 “在线用户” 的列表。当再有新用户连接上来时，应向该列表中所有用户进行广播通知，提示 xxx 用户上线。
当然，简单实现手法可以循环读取列表中的用户，依次向其发送消息通知新用户上线。但这种方式无疑是一种串行的通信手段，实现简单，但执行效率较低。
在 go 语言中，我们利用 go 程轻便、高效、并发性好的特性，给每个登录用户维护多个 go 程来进行数据通信，借助 channel 不需要使用同步锁，就可以实现高效的并发通信。
下图充分利用 goroutine 和 channel 实现了新用户登录，向所有在线用户进行广播通知：
分析上图，主要分为几大模块。
全局位置定义用户结构体类型 Client，存储登录用户信息。成员包含 channel、Name、Addr
type Client struct {
C chan string
Name string
Addr string
}
定义全局通道 message 处理消息。
定义全局 map 存储在线用户信息。Key 为用户网络地址。Value 为用户结构体。
主 go 程，监听客户端连接请求，当有新的客户端连接，创建新 go 程 handleConnet 处理用户连接。
handleConnet go 程，获取用户网络地址（Ip+port），创建新用户结构体，包含成员 C、Name、Addr。新用户的 Name 和 Addr 初值都是用户网络地址（Ip+port）。将用户结构体存入 map 中。并创建 WriteMsgToClient go 程，专门负责给当前用户发送消息。组织新用户上线广播消息内容，写入全局通道 message 中。
WriteMsgToClient go 程，读取用户结构体 C 中的数据，没有则阻塞等待，有数据写出给登录用户。
Manager go 程，给 map 分配空间。循环读取 message 通道中是否有数据。没有，阻塞等待。有则解除阻塞，将 message 通道中读到的数据写到用户结构体中的 C 通道。
代码实现：
package main
import (
"net"
"fmt"
)
// 定义用户结构体类型
type Client struct {
C chan string
Name string
Addr string
}
// 定义全局 map 存储在线用户 key:IP+port, value:Client
var onlineMap map[string]Client
// 定义全局 channel 处理消息
var message = make(chan string)
func WriteMsgToClient(clnt Client, conn net.Conn) {
// 循环跟踪 clnt.C，有消息则读走，Write 给客户端
for msg := range clnt.C {
conn.Write([]byte(msg + "\n")) // 发送消息 给客户端
}
}
func MakeMsg(clnt Client, msg string) (buf string) {
buf = "[" + clnt.Addr + "]" + clnt.Name + ": " + msg
return
}
func HandleConnect(conn net.Conn) {
defer conn.Close()
// 获取新连接上来的用户的网络地址(IP+port)
netAddr := conn.RemoteAddr().String()
// 给新用户创建结构体。用户名、网络地址一样
clnt := Client{make(chan string), netAddr, netAddr}
// 将新创建的结构体，添加到 map 中，key值为获取到的网络地址（IP+port）
onlineMap[netAddr] = clnt
// 新创建一个go程，专门给当前客户端发送消息。
go WriteMsgToClient(clnt, conn)
// 广播新用户上线
// message <- "[" + clnt.Addr + "]" + clnt.Name + ": login"
message <- MakeMsg(clnt, "login")
for { // 防止当前go程结束。
runtime.GC()
}
}
func Manager() {
// 给map分配空间
onlineMap = make(map[string]Client)
// 循环读取 message 通道中的数据
for {
// 通道 message 中有数据读到 msg 中。 没有，则阻塞
msg := <-message
// 一旦执行到这里，说明message中有数据了，解除阻塞。 遍历 map
for _, clnt := range onlineMap {
clnt.C <- msg // 把从Message通道中读到的数据，写到 client 的 C 通道中。
}
}
}
func main() {
// 创建监听 socket
listener, err := net.Listen("tcp", "127.0.0.1: 8000")
if err != nil {
fmt.Println("Listen err:", err)
return
}
defer listener.Close()
// 创建go程 处理消息
go Manager()
// 循环接收客户端连接请求
for {
conn, err := listener.Accept()
if err != nil {
fmt.Println("Accept err:", err)
continue // 失败，监听其他客户端连接
}
// 给新连接的客户端，单独创建一个go程，处理客户端连接请求
go HandleConnect(conn)
}
}
广播用户消息#
当某个客户端向服务端发送消息后，服务端应将该消息广播给其它的客户端，达到聊天室的群聊效果。
开启一个新的 go 程，为方便传参，可以选择匿名 go 程。专门负责接收从客户端传递过来的数据，然后将接收到的数据写到 messaage 通道中。
在实现 “广播用户上线” 时，我已经完成：Manager go 程会阻塞读 message 通道，一旦有数据，则遍历 map 中的在线用户。将数据写到结构体成员的 C 通道中。WriteMsgToClient go 程会迭代 C 这个 channel，最终将数据发送给客户端。
综上，实际上我们想完成 “广播用户消息” 给所有在线用户的功能，只需要将读到的数据写到 message 通道即可达到目的。
相关代码：
func HandleConnect(conn net.Conn) {
……
……
// 广播新用户上线
message <- MakeMsg(clnt, "login")
// 创建一个新go程，循环读取用户发送的消息，广播给在线用户
go func() {
buf := make([]byte, 2048) // 定义切片缓冲区，存储读到的用户消息
for {
n, err := conn.Read(buf)
if n == 0 { // 用户退出登录
fmt.Printf("用户%s退出登录\n", clnt.Name)
return
}
if err != nil {
fmt.Println("Read err:", err)
return
}
msg := string(buf[:n]) // 保存用户写来的消息内容
message <-MakeMsg(clnt, msg) // 将消息广播给所有在线用户
}
}()
for { // 不能让当前go程结束。
;
}
}
展示在线用户#
因为 nc 工具默认会添加‘\n’， 所以 conn.Read () 读取用户消息后，修改保存用户消息内容实现语句：
msg := string (buf [:n-1]) 重新读取用户消息。
读到后，对消息内容进行判断：如果用户发送了 “who”，则当成一个查询指令处理。遍历 map 中所有在线用户，取出每个用户的相关描述信息，组成提示消息，写给当前用户即可。
由于这里客户端我们使用 nc 工具模拟，该工具对中文支持较差，所以我们组织的用户描述信息中不要包含中文字符。
代码片段如下：
msg := string(buf[:n-1]) // 保存用户写来的消息内容, nc 工具默认添加‘\n’
if msg == "who" && len(msg) == 3 { // 判断用户发送了 who 指令
conn.Write([]byte("user list:\n"))
for _, user := range onlineMap { // 遍历map获取在线用户
userInfo := user.Addr + ":" + user.Name + "\n" // 组织在线用户信息
conn.Write([]byte(userInfo)) // 写给当前用户
}
} else {
message <-MakeMsg(clnt, msg) // 将消息广播给所有在线用户
}
修改用户名#
前面我们查看用户信息时，用户名都是与用户网络地址相同的内容。主要由于用户登录时，创建该用户名不是用户自己完成的，无法洞悉用户的意图。当用户成功登录上来可以通过给服务器发送消息，来修改自己的用户名。
设定，如果用户发送 “rename | Iron man” 指令，既是想修改自己的用户名为 “Iron man”。判断用户消息，是否包含 “rename|” 关键字：if len (msg) >= 8 && msg [:6] == "rename" {。如果是，那么拆分用户意欲修改的用户名保存。strings.Split () 函数可以完成拆分字符串操作。
将该用户名替换当前用户的 Name。使用用户的 Addr 作为 key，找到 map 中当前用户，覆盖即可达到改名的目的。操作结束提示用户改名成功。
代码片段如下：
msg := string(buf[:n-1])
if msg == "who" && len(msg) == 3 {
conn.Write([]byte("user list:\n"))
for _, user := range onlineMap {
userInfo := user.Addr + ":" + user.Name + "\n"
conn.Write([]byte(userInfo))
}
// 判断用户输入的前6个字符是否为 rename
} else if len(msg) >= 8 && msg[:6] == "rename" { // rename | Iron man
newName := strings.Split(msg, "|")[1] // 按"|"拆分，rename为[0], Iron man为[1]
clnt.Name = newName // 替换掉当前用户原始Name
onlineMap[netAddr] = clnt // 使用netAddr为key找到map中当前用户。覆盖
conn.Write([]byte("rename successful\n"))
} else {
message <- MakeMsg(clnt, msg)
}
用户退出#
前面在 “广播用户消息” 时，当 conn.Read () 读到 0 时，我们在服务器端，简单打印了 “用户 xxx 退出登录” 的提示。
但实际上，在聊天室中，有在线用户离开，我们应该将这一事件广播给所有用户知晓，并且将该用户从 map 在线用列表中移除。需要实时的监看在线用户的状态。可以创建 channel 来检测用户退出状态，并使用 select 来监听 channel 上的数据流动。
当 channel 上有数据时，select 对应阻塞 case 语句得以执行。将用户从 map 中移除。同时通知所有在线用户。
代码片段：
func HandleConnect(conn net.Conn) {
……
message <- MakeMsg(clnt, "login")
isQuit := make(chan bool) // 检测用户主动退出
go func() {
buf := make([]byte, 2048)
for {
n, err := conn.Read(buf)
if n == 0 {
isQuit <- true // 用户主动退出登录
fmt.Printf("用户%s退出登录\n", clnt.Name)
return
}
……
}
}()
for {
select {
case <-isQuit: // 用户不主动退出，阻塞
close(clnt.C)
delete(onlineMap, netAddr) // 将当前用户从map中移除
message <- MakeMsg(clnt, "logout") // 广播给在线用户，谁退出了
return // 结束当前退出用户对应go程
}
}
}
超时处理#
如果客户端没有主动退出，并且长时间没有发送消息，会一直占用服务端的资源。服务器通常针对这种用户添加 “超时强踢” 机制，强制将该客户端与服务器连接断开。
可以借助并发编程时我们所学的 select 超时机制来实现。Select 监听 time.After (60 * time.Second) 通道上的数据流动。如果在计时期间一直没有数据，通道中会被写入当前系统时间，select 的 case 满足读条件，不再阻塞。但，有一个问题，用户如果持续在输入数据，这个计时器依然在计时，时间到，依然会强制踢出用户。
因此，我们另外创建一个通道 hasData 来检测用户是否有数据发送，让 Select 也来监听这个 channel。这样，当用户有数据输入时，select 监听的这个 hasData 通道会满足 case 条件得以执行，但我们不做任何处理。目的是使得监听在 select 中的计时器被重新计时。
只有当真正持续 60s 没有数据发送时，select 中用于计时的 case 才满足条件，将用户与服务器连接断开。
代码片段：
func HandleConnect(conn net.Conn) {
……
……
isQuit := make(chan bool)
hasData := make(chan bool) // 检测用户是否有消息发送
go func() {
buf := make([]byte, 2048)
for {
n, err := conn.Read(buf)
……
msg := string(buf[:n-1])
if {
……
} else if {
……
} else {
……
}
hasData <- true // 只要执行到这里，就说明用户有数据发送
}
}()
for {
select {
case <-isQuit:
close(clnt.C)
delete(onlineMap, netAddr)
message <- MakeMsg(clnt, "logout")
return
case <-hasData:
// 什么都不做，目的是让计时器归零
case <-time.After(60*time.Second):
close(clnt.C)
delete(onlineMap, netAddr) // 将当前用户从map中移除
message <- MakeMsg(clnt, "time out leave") // 广播给在线用户，超时退出
return // 结束当前退出用户对应go程
}
}
}
这里需要注意的是，每循环一次，第三个 case 后面的时间都会重新计算。（例如：执行完 case<-hasData 后，紧跟着执行第三个 case，发现时间是 10 秒，不到 60 秒，条件不成立，不会执行该 case 后面的代码，进入下次循环，这时时间重新计算）
当 hasData 没有数据，isQuit 没有数据，60s 时间没有到，这时三个 case 都阻塞等待。直到 60 秒后，前两个 case 条件依然不成立，第三个 case 满足，执行后面代码，断开客户端连接，踢下线。
关于 LearnKu
粤公网安备 44030502004330号
推荐文章：