refactor(tcp): 重构TCP消息处理机制

- 修改导入路径从gf-common到gin-base
- 实现消息帧解析功能,支持4字节长度前缀协议
- 添加连接缓冲区机制处理分包粘包问题
- 移除旧的消息处理逻辑并替换为新的帧解析方法
- 更新消息发送逻辑以使用帧格式传输数据
- 调整GOHCache配置文件以匹配新的代码结构
This commit is contained in:
2026-03-02 09:25:43 +08:00
parent 2520655fbd
commit 86116618e4
3 changed files with 107 additions and 242 deletions
+63 -24
View File
@@ -2,11 +2,12 @@ package tcp
import (
"context"
"encoding/binary"
"fmt"
"sync"
"time"
"git.magicany.cc/black1552/gf-common/pool"
"git.magicany.cc/black1552/gin-base/pool"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/net/gtcp"
"github.com/gogf/gf/v2/os/glog"
@@ -181,6 +182,8 @@ func (s *TCPServer) receiveMessages(conn *TcpConnection) {
now := time.Now()
conn.Mutex.Lock()
conn.LastUsed = now
// 将读取的数据添加到连接的缓冲区
conn.buffer = append(conn.buffer, buffer[:n]...)
conn.Mutex.Unlock()
// 更新SQLite中的连接信息
@@ -193,33 +196,62 @@ func (s *TCPServer) receiveMessages(conn *TcpConnection) {
}
}
// 处理消息
data := make([]byte, n)
copy(data, buffer[:n])
msg := &TcpMessage{
Id: fmt.Sprintf("msg_%d", gtime.TimestampNano()),
ConnId: conn.Id,
Data: data,
Timestamp: time.Now(),
IsSend: false,
}
// 使用协程池处理消息,避免阻塞
grpool.AddWithRecover(s.ctx, func(ctx context.Context) {
if s.MessageHandler != nil {
if err := s.MessageHandler(conn, msg); err != nil {
s.Logger.Error(s.ctx, fmt.Sprintf("Message handling error: %v", err))
}
}
}, func(ctx context.Context, err error) {
s.Logger.Error(ctx, fmt.Sprintf("Message handling error: %v", err))
})
// 解析消息帧
s.parseMessageFrames(conn)
}
}
}
}
// parseMessageFrames 解析消息帧
func (s *TCPServer) parseMessageFrames(conn *TcpConnection) {
conn.Mutex.Lock()
defer conn.Mutex.Unlock()
for {
// 检查缓冲区是否有足够的数据来读取长度前缀
if len(conn.buffer) < messageLengthPrefixSize {
// 数据不足,等待下一次读取
return
}
// 读取长度前缀
length := binary.BigEndian.Uint32(conn.buffer[:messageLengthPrefixSize])
// 检查缓冲区是否有足够的数据来读取完整的消息
if len(conn.buffer) < messageLengthPrefixSize+int(length) {
// 数据不足,等待下一次读取
return
}
// 提取消息数据
data := conn.buffer[messageLengthPrefixSize : messageLengthPrefixSize+int(length)]
// 移除已处理的消息数据
conn.buffer = conn.buffer[messageLengthPrefixSize+int(length):]
// 创建消息对象
msg := &TcpMessage{
Id: fmt.Sprintf("msg_%d", gtime.TimestampNano()),
ConnId: conn.Id,
Data: data,
Timestamp: time.Now(),
IsSend: false,
}
// 使用协程池处理消息,避免阻塞
grpool.AddWithRecover(s.ctx, func(ctx context.Context) {
if s.MessageHandler != nil {
if err := s.MessageHandler(conn, msg); err != nil {
s.Logger.Error(s.ctx, fmt.Sprintf("Message handling error: %v", err))
}
}
}, func(ctx context.Context, err error) {
s.Logger.Error(ctx, fmt.Sprintf("Message handling error: %v", err))
})
}
}
// SendTo 发送消息到指定连接
func (s *TCPServer) SendTo(connID string, data []byte) error {
conn := s.Connection.Get(connID)
@@ -249,8 +281,15 @@ func (s *TCPServer) sendMessage(conn *TcpConnection, data []byte) error {
// 设置写入超时
conn.Server.SetWriteDeadline(time.Now().Add(s.Config.WriteTimeout))
// 创建消息帧:4字节长度前缀 + 消息数据
frame := make([]byte, messageLengthPrefixSize+len(data))
// 写入长度前缀(大端序)
binary.BigEndian.PutUint32(frame[:messageLengthPrefixSize], uint32(len(data)))
// 写入消息数据
copy(frame[messageLengthPrefixSize:], data)
// 发送数据
_, err := conn.Server.Write(data)
_, err := conn.Server.Write(frame)
if err != nil {
return err
}
+6
View File
@@ -7,6 +7,11 @@ import (
"github.com/gogf/gf/v2/net/gtcp"
)
// 消息帧格式:4字节长度前缀 + 消息数据
const (
messageLengthPrefixSize = 4 // 消息长度前缀大小(4字节)
)
// TcpPoolConfig TCP连接池配置
type TcpPoolConfig struct {
BufferSize int `json:"bufferSize"` // 缓冲区大小
@@ -26,6 +31,7 @@ type TcpConnection struct {
LastUsed time.Time `json:"lastUsed"` // 最后使用时间
CreatedAt time.Time `json:"createdAt"` // 创建时间
Mutex sync.RWMutex `json:"-"` // 读写锁
buffer []byte `json:"-"` // 用于存储未处理的字节数据
}
// TcpMessage TCP消息结构