基于Go语言实现一个网络谈天室（连接Redis版） - IT评测·应用市场-qidao123.com技术社区

conn, err := net.Dial("tcp", "localhost:8080")
if err != nil {
log.Fatal("连接服务器时出错:", err)
}
defer conn.Close()

复制代码

fmt.Print("请输入你的网名: ")
nameInput, err := reader.ReadString('\n')
name = strings.TrimSpace(nameInput)
err = utils.SendMessage(conn, []byte(name))

复制代码

message, err := reader.ReadString('\n')
err = utils.SendMessage(conn, []byte(message))

复制代码

func handleServerMessages(conn *net.Conn) {
reader := bufio.NewReader(*conn)
for {
message, err := utils.ReadMessage(reader)
if err != nil {
// 处理错误和重连逻辑
}
fmt.Println(string(message))
}
}

复制代码

if string(message) == "PING" {
log.Printf("【心跳检测】接收到服务端 PING，发送 PONG 响应")
utils.SendMessage(*conn, []byte("PONG"))
}

复制代码

func reconnect(oldConn net.Conn) (net.Conn, bool) {
for i := 0; i < 3; i++ {
newConn, err := net.Dial("tcp", "localhost:8080")
if err == nil {
oldConn.Close()
return newConn, true
}
time.Sleep(time.Duration(2<<uint(i)) * time.Second)
}
return nil, false
}

复制代码

redisClient := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "",
DB: 0,
})

复制代码

type ChatRoom struct {
Clients map[*Client]bool
Join chan *Client
Leave chan *Client
Message chan []byte
Redis *redis.Client
mu sync.RWMutex
}

复制代码

func (cr *ChatRoom) Run() {
for {
select {
case message := <-cr.Message:
cr.mu.RLock()
for client := range cr.Clients {
client.Send <- message
}
cr.mu.RUnlock()
batchStoreToRedis(ctx, cr.Redis, message)
}
}
}

复制代码

ticker := time.NewTicker(10 * time.Second)
heartbeatTimeout := time.NewTimer(15 * time.Second)
for {
select {
case <-ticker.C:
utils.SendMessage(client.Conn, []byte("PING"))
heartbeatTimeout.Reset(15 * time.Second)
case <-heartbeatTimeout.C:
cr.Leave <- client
}
}

复制代码

func batchStoreToRedis(ctx context.Context, redisClient *redis.Client, message []byte) {
queue := redisClient.LRange(ctx, "chat_messages_queue", 0, -1).Val()
queue = append(queue, string(message))
if len(queue) >= 10 {
pipe := redisClient.Pipeline()
for _, msg := range queue {
pipe.RPush(ctx, "chat_messages", msg)
pipe.ZIncrBy(ctx, "user_activity", 1, extractUsername(msg))
}
pipe.Exec(ctx)
redisClient.Del(ctx, "chat_messages_queue")
} else {
redisClient.RPush(ctx, "chat_messages_queue", string(message))
}
}

复制代码

func SendMessage(conn net.Conn, message []byte) error {
length := uint32(len(message))
err := binary.Write(conn, binary.BigEndian, length)
if err != nil {
return err
}
_, err = conn.Write(message)
return err
}
func ReadMessage(reader *bufio.Reader) ([]byte, error) {
var length uint32
err := binary.Read(reader, binary.BigEndian, &length)
if err != nil {
return nil, err
}
message := make([]byte, length)
_, err = io.ReadFull(reader, message)
return message, err
}

复制代码

for {
select {
case <-ticker.C: // 心跳检测分支（每10秒触发一次）
sendPing()
default: // 默认分支（非阻塞）
message := readMessage()
processMessage(message)
}
}

复制代码

时间点 0ms: 服务端开始循环
时间点 5ms: 客户端发送消息A
时间点 10ms: ticker.C 触发心跳检测（发送PING）
时间点 15ms: 客户端发送消息B

复制代码

for {
select {
case <-ticker.C: // 心跳检测
sendPing()
case <-msgTicker.C: // 专用消息读取分支（每50ms触发一次）
message := readMessage()
processMessage(message)
}
}

复制代码

时间点 0ms: 用户A发送网名“小明” → 服务端接受并加入。
时间点 50ms: 用户B发送网名“小明” → 服务端未检查唯一性，直接加入。
时间点 100ms: 用户A发送消息“你好”，用户B发送消息“大家好”。

复制代码

exists, err := cr.Redis.SIsMember(ctx, "online_users", clientName).Result()
if exists {
utils.SendMessage(conn, []byte("ERROR: 网名已被占用"))
return
}

复制代码

cr.Redis.SAdd(ctx, "online_users", clientName)

复制代码

defer cr.Redis.SRem(ctx, "online_users", clientName)

复制代码

时间点 0ms: 用户输入空网名（直接回车） → 服务端允许加入。
时间点 100ms: 用户发送空消息（空格 + 回车） → 服务端广播“: ”。

复制代码

for {
fmt.Print("请输入网名: ")
name = strings.TrimSpace(reader.ReadString())
if name != "" {
break
}
fmt.Println("网名不能为空！")
}

复制代码

message = strings.TrimSpace(input)
if message == "" {
fmt.Println("消息不能为空！")
continue
}

复制代码

if clientName == "" {
utils.SendMessage(conn, []byte("ERROR: 网名不能为空"))
return
}

复制代码

msgContent := strings.TrimSpace(string(message))
if msgContent == "" {
log.Printf("客户端 %s 发送了空消息，已忽略", client.Name)
continue
}

复制代码

时间点 0ms: 服务端发送PING。
时间点 10ms: 客户端未响应（已断开）。
时间点 25ms: 服务端仍认为客户端在线，未清理资源。

复制代码

ticker := time.NewTicker(10 * time.Second)
defer ticker.Stop()

复制代码

case <-msgTicker.C:
if message == "PONG" {
heartbeatTimeout.Reset(15 * time.Second)
}

复制代码

heartbeatTimeout := time.NewTimer(15 * time.Second)
defer heartbeatTimeout.Stop()
select {
case <-heartbeatTimeout.C:
cr.Leave <- client
return
}

复制代码

package main
import (
"bufio"
"fmt"
"log"
"net"
"os"
"strings"
"time"
"Learn/kaohe/redis2/utils"
)
// handleServerMessages 负责处理从服务器接收到的消息
func handleServerMessages(conn *net.Conn) {
reader := bufio.NewReader(*conn) // 创建一个读取器，用于从连接中读取消息
for {
message, err := utils.ReadMessage(reader) // 从服务器读取消息
if err != nil {
log.Println("接收服务器消息时出错:", err)
// 如果连接断开，尝试重新连接
newConn, ok := reconnect(*conn)
if !ok {
return // 如果重新连接失败，退出处理
}
*conn = newConn // 更新连接
reader = bufio.NewReader(*conn) // 重新创建读取器
continue
}
log.Printf("接收到消息: %s", string(message))
fmt.Println(string(message)) // 打印消息到控制台
// 如果收到的是 PING 消息，发送 PONG 响应
if string(message) == "PING" {
log.Printf("【心跳检测】接收到服务端 PING，发送 PONG 响应")
err := utils.SendMessage(*conn, []byte("PONG"))
if err != nil {
log.Println("发送心跳响应时出错:", err)
}
}
}
}
// reconnect 尝试重新连接到服务器
func reconnect(oldConn net.Conn) (net.Conn, bool) {
for i := 0; i < 3; i++ { // 最多尝试 3 次
log.Printf("尝试第 %d 次重新连接服务器...", i+1)
newConn, err := net.Dial("tcp", "localhost:8080") // 尝试连接服务器
if err == nil {
log.Println("重新连接服务器成功")
oldConn.Close() // 关闭旧连接
return newConn, true // 返回新连接
}
time.Sleep(time.Duration(2<<uint(i)) * time.Second) // 指数退避等待
}
log.Println("多次尝试重新连接服务器失败，退出程序")
return nil, false // 重新连接失败
}
func main() {
conn, err := net.Dial("tcp", "localhost:8080") // 连接到服务器
if err != nil {
log.Fatal("连接服务器时出错:", err)
}
defer conn.Close() // 程序结束时关闭连接
go handleServerMessages(&conn) // 启动一个 goroutine 处理服务器消息
reader := bufio.NewReader(os.Stdin) // 创建一个读取器，用于读取用户输入
// 循环读取网名直到输入有效
var name string
for {
fmt.Print("请输入你的网名: ")
nameInput, err := reader.ReadString('\n') // 读取用户输入的网名
if err != nil {
log.Fatal("读取网名时出错:", err)
}
name = strings.TrimSpace(nameInput) // 去除多余空格
if name == "" {
fmt.Println("网名不能为空，请重新输入！")
continue
}
break
}
err = utils.SendMessage(conn, []byte(name)) // 向服务器发送网名
if err != nil {
log.Fatal("发送网名时出错:", err)
}
fmt.Println("你已成功加入聊天室，可以开始聊天了！")
for {
message, err := reader.ReadString('\n') // 读取用户输入的消息
if err != nil {
log.Println("读取用户输入时出错:", err)
break
}
message = strings.TrimSpace(message) // 去除多余空格
if message == "" {
fmt.Println("消息不能为空，请重新输入！")
continue
}
if message == "/quit" { // 如果输入是 "/quit"，退出聊天室
fmt.Println("你已退出聊天室")
utils.SendMessage(conn, []byte("/quit"))
break
}
err = utils.SendMessage(conn, []byte(message)) // 向服务器发送消息
log.Printf("DEBUG: 客户端尝试发送消息: %s, 错误: %v", message, err)
if err != nil {
log.Println("发送消息到服务器时出错:", err)
// 如果发送失败，尝试重新连接
newConn, ok := reconnect(conn)
if !ok {
return
}
conn = newConn
err = utils.SendMessage(conn, []byte(name)) // 重新发送网名
if err != nil {
log.Fatal("重新连接后发送网名时出错:", err)
}
}
}
}

复制代码

package mainimport ( "bufio" "context" "fmt" "io" "log" "net" "strings" "sync" "time" "Learn/kaohe/redis2/utils" "github.com/redis/go-redis/v9")// Client 表示一个客户端连接type Client struct { Name string // 客户端的网名 Conn net.Conn // 客户端的连接 Send chan []byte // 用于向客户端发送消息的通道}// ChatRoom 表示谈天室，管理所有客户端连接type ChatRoom struct { Clients map[*Client]bool // 当前在线的客户端 Join chan *Client // 新客户端加入的通道 Leave chan *Client // 客户端离开的通道 Message chan []byte // 广播消息的通道 Redis *redis.Client // Redis 客户端 mu sync.RWMutex // 用于并发控制的互斥锁}// NewChatRoom 创建一个新的谈天室func NewChatRoom(redisClient *redis.Client) *ChatRoom { return &ChatRoom{ Clients: make(map[*Client]bool), // 初始化客户端列表 Join: make(chan *Client), // 初始化加入通道 Leave: make(chan *Client), // 初始化离开通道 Message: make(chan []byte, 100), // 初始化消息通道，缓冲大小为 100 Redis: redisClient, // 初始化 Redis 客户端 }}// Run 谈天室的主循环，处理客户端的加入、离开和消息广播func (cr *ChatRoom) Run() { for { select { case client := <-cr.Join: // 处理新客户端加入 cr.mu.Lock() cr.Clients[client] = true // 将客户端添加到列表 cr.mu.Unlock() message := fmt.Sprintf("系统广播: %s 加入了谈天室", client.Name) cr.Message <- []byte(message) // 广播加入消息 case client := <-cr.Leave: // 处理客户端离开 cr.mu.Lock() if _, ok := cr.Clients[client]; ok { delete(cr.Clients, client) // 从列表中删除客户端 close(client.Send) // 关闭客户端的发送通道 } cr.mu.Unlock() if client.Name != "" { message := fmt.Sprintf("系统广播: %s 离开了谈天室", client.Name) cr.Message <- []byte(message) // 广播离开消息 } case message := <-cr.Message: // 处理广播消息 cr.mu.RLock() log.Printf("准备广播消息: %s", string(message)) for client := range cr.Clients { // 向所有客户端发送消息 select { case client.Send <- message: // 发送消息 log.Printf("已发送消息到客户端 %s", client.Name) default: log.Printf("客户端 %s 的发送通道已满", client.Name) } } cr.mu.RUnlock() // 将消息存储到 Redis ctx := context.Background() batchStoreToRedis(ctx, cr.Redis, message) } }}// batchStoreToRedis 将消息批量存储到 Redisfunc batchStoreToRedis(ctx context.Context, redisClient *redis.Client, message []byte) { const batchSize = 10 // 批量存储的大小 // 从 Redis 获取当前队列中的消息 queue := redisClient.LRange(ctx, "chat_messages_queue", 0, -1).Val() queue = append(queue, string(message)) // 将新消息添加到队列 if len(queue) >= batchSize { // 如果队列到达批量大小 pipe := redisClient.Pipeline() // 创建 Redis 管道 for _, msg := range queue { // 将消息存储到 chat_messages 列表 pipe.RPush(ctx, "chat_messages", msg) // 更新用户活跃度 pipe.ZIncrBy(ctx, "user_activity", 1, extractUsername(msg)) } _, _ = pipe.Exec(ctx) // 执行管道 redisClient.Del(ctx, "chat_messages_queue") // 删除队列 } else { // 如果队列未到达批量大小，将消息存储到队列 redisClient.RPush(ctx, "chat_messages_queue", string(message)) }}// extractUsername 从消息中提取用户名func extractUsername(message string) string { parts := strings.SplitN(message, ": ", 2) if len(parts) > 0 { return parts[0] } return ""}// HandleClient 处理单个客户端的连接func (cr *ChatRoom) HandleClient(conn net.Conn) { var clientName string defer func() { conn.Close() // 关闭连接 cr.Leave <- &Client{Name: clientName, Conn: conn, Send: nil} // 关照谈天室客户端离开 log.Printf("客户端连接已关闭: %s", conn.RemoteAddr()) }() reader := bufio.NewReader(conn) // 创建读取器 nameBytes, err := utils.ReadMessage(reader) // 读取客户端的网名 if err != nil { log.Println("读取网名时出错:", err) return } clientName = strings.TrimSpace(string(nameBytes)) // 检查网名是否为空 if clientName == "" { log.Println("客户端尝试使用空网名连接") utils.SendMessage(conn, []byte("ERROR: 网名不能为空")) return } // 检查网名是否已存在 ctx := context.Background() exists, err := cr.Redis.SIsMember(ctx, "online_users", clientName).Result() if err != nil { log.Printf("检查网名 %s 时 Redis 出错: %v", clientName, err) utils.SendMessage(conn, []byte("ERROR: 服务器内部错误")) return } if exists { log.Printf("网名 %s 已存在，拒绝连接", clientName) utils.SendMessage(conn, []byte("ERROR: 网名已被占用，请更换其他名称")) return } // 添加网名到在线用户聚集 if _, err := cr.Redis.SAdd(ctx, "online_users", clientName)
.Result(); err != nil { log.Printf("存储网名 %s 到 Redis 失败: %v", clientName, err) utils.SendMessage(conn, []byte("ERROR: 服务器内部错误")) return } defer cr.Redis.SRem(ctx, "online_users", clientName)
client := &Client{ Name: clientName, Conn: conn, Send: make(chan []byte, 10), // 初始化发送通道 } cr.Join <- client // 关照谈天室客户端加入 // 启动一个 goroutine 处理发送消息 go func() { for message := range client.Send { err := utils.SendMessage(client.Conn, message) if err != nil { log.Println("发送消息给客户端时出错:", err) cr.Leave <- client // 关照谈天室客户端离开 break } } }() // 定时发送心跳检测 ticker := time.NewTicker(10 * time.Second) // 定时读取消息 msgTicker := time.NewTicker(50 * time.Millisecond) // 心跳超时检测 heartbeatTimeout := time.NewTimer(15 * time.Second) defer func() { ticker.Stop() msgTicker.Stop() heartbeatTimeout.Stop() }() for { select { case <-ticker.C: // 心跳检测 log.Printf("【心跳检测】向客户端 %s 发送 PING", client.Name) if err := utils.SendMessage(client.Conn, []byte("PING")); err != nil { log.Printf("【心跳非常】客户端 %s 心跳发送失败: %v", client.Name, err) cr.Leave <- client return } heartbeatTimeout.Reset(15 * time.Second) case <-heartbeatTimeout.C: log.Printf("【心跳超时】客户端 %s 未响应 PONG，逼迫断开", client.Name) cr.Leave <- client return case <-msgTicker.C: // 读取消息 message, err := utils.ReadMessage(reader) if err != nil { if err == io.EOF { log.Printf("客户端 %s 自动断开连接", client.Name) } else { log.Printf("读取消息时出错: %v", err) } cr.Leave <- client return } if string(message) == "PONG" { // 处理心跳响应 log.Printf("【心跳响应】客户端 %s 返回 PONG", client.Name) heartbeatTimeout.Reset(15 * time.Second) continue } msgContent := strings.TrimSpace(string(message)) if msgContent == "" { log.Printf("客户端 %s 发送了空消息，已忽略", client.Name) continue } log.Printf("接收到来自 %s 的消息: %s", client.Name, msgContent) // 构造完整消息并广播 fullMessage := fmt.Sprintf("%s: %s", client.Name, msgContent) cr.Message <- []byte(fullMessage) } }}// printActivityRankings 打印用户活跃度排名func printActivityRankings(redisClient *redis.Client) { ctx := context.Background() topUsers, err := redisClient.ZRevRangeWithScores(ctx, "user_activity", 0, 9).Result() if err != nil { log.Println("获取活跃度排名时出错:", err) return } fmt.Println("活跃度排名:") for i, user := range topUsers { fmt.Printf("%d. %s: %.0f\n", i+1, user.Member.(string), user.Score) }}// PrintConnectedClients 打印当前在线客户端func (cr *ChatRoom) PrintConnectedClients() { cr.mu.RLock() defer cr.mu.RUnlock() log.Println("【活跃客户端】当前在线客户端列表:") for client := range cr.Clients { log.Printf(" - %s (IP: %s)", client.Name, client.Conn.RemoteAddr()) }}func main() { // 初始化 Redis 客户端 redisClient := redis.NewClient(&redis.Options{ Addr: "localhost:6379", Password: "", DB: 0, }) ctx := context.Background() _, err := redisClient.Ping(ctx).Result() if err != nil { log.Fatal("连接 Redis 时出错:", err) } // 监听端口 listener, err := net.Listen("tcp", ":8080") if err != nil { log.Fatal("监听端口时出错:", err) } defer listener.Close() fmt.Println("谈天室服务器已启动，等待客户端连接...") // 创建谈天室 chatRoom := NewChatRoom(redisClient) go chatRoom.Run() // 启动谈天室 // 定期打印活跃度排名和在线客户端 go func() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for range ticker.C { printActivityRankings(redisClient) chatRoom.PrintConnectedClients() } }() // 担当客户端连接 for { conn, err := listener.Accept() if err != nil { log.Println("担当客户端连接时出错:", err) continue } go chatRoom.HandleClient(conn) // 处理客户端连接 }}

复制代码

package utils
import (
"bufio"
"encoding/binary"
"fmt"
"io"
"log"
"net"
)
const maxMessageLength = 1 << 20 // 1MB，最大消息长度限制
// SendMessage 向连接发送消息
func SendMessage(conn net.Conn, message []byte) error {
length := uint32(len(message)) // 消息长度
if length > maxMessageLength {
log.Println("消息长度超出限制:", length)
return fmt.Errorf("message too long")
}
// 写入消息长度
err := binary.Write(conn, binary.BigEndian, length)
if err != nil {
log.Printf("写入消息长度时出错: %v", err)
return err
}
// 写入消息内容
_, err = conn.Write(message)
if err != nil {
log.Printf("写入消息内容时出错: %v", err)
}
return err
}
// ReadMessage 从连接读取消息
func ReadMessage(reader *bufio.Reader) ([]byte, error) {
var length uint32 // 消息长度
err := binary.Read(reader, binary.BigEndian, &length)
if err != nil {
log.Printf("读取消息长度时出错: %v", err)
return nil, err
}
if length > maxMessageLength {
log.Println("消息长度超出限制:", length)
return nil, fmt.Errorf("message too long")
}
// 读取消息内容
message := make([]byte, length)
_, err = io.ReadFull(reader, message)
if err != nil {
log.Printf("读取消息内容时出错: %v", err)
return nil, err
}
return message, nil
}

复制代码