236 lines
6.1 KiB
Markdown
236 lines
6.1 KiB
Markdown
# 高性能网络通信框架
|
||
|
||
一个用 Go 语言编写的高性能、高并发网络通信框架，支持 TCP 和 UDP 协议。该框架专为高吞吐量和低延迟的网络应用设计，适用于游戏服务器、聊天系统、实时数据处理等场景。
|
||
|
||
## 主要特性
|
||
|
||
- **协议支持**：同时支持 TCP 和 UDP 协议
|
||
- **高并发处理**：使用工作协程池高效处理请求
|
||
- **连接分片管理**：通过分片减少锁竞争，提高并发性能
|
||
- **自动连接生命周期管理**：自动处理连接的创建、维护和关闭
|
||
- **内存优化**：使用缓冲区池和对象池减少内存分配和 GC 压力
|
||
- **优雅关闭**：支持服务器的平滑关闭，确保数据不丢失
|
||
- **超时控制**：支持读写超时和空闲连接超时设置
|
||
- **可定制的数据包处理**：灵活的数据包接口，支持自定义协议
|
||
|
||
## 安装
|
||
|
||
```bash
|
||
go get github.com/yourusername/network
|
||
```
|
||
|
||
## 快速开始
|
||
|
||
### TCP 服务器示例
|
||
|
||
```go
|
||
package main
|
||
|
||
import (
|
||
"log"
|
||
"time"
|
||
|
||
"github.com/yourusername/network"
|
||
)
|
||
|
||
// 自定义数据包
|
||
type MyPacket struct {
|
||
Type byte
|
||
Length uint16
|
||
Payload []byte
|
||
}
|
||
|
||
// 实现 Packet 接口
|
||
func (p *MyPacket) Decode(data []byte) error {
|
||
if len(data) < 3 {
|
||
return errors.New("invalid packet")
|
||
}
|
||
p.Type = data[0]
|
||
p.Length = uint16(data[1])<<8 | uint16(data[2])
|
||
if len(data) < int(3+p.Length) {
|
||
return errors.New("incomplete packet")
|
||
}
|
||
p.Payload = data[3:3+p.Length]
|
||
return nil
|
||
}
|
||
|
||
func (p *MyPacket) Encode() ([]byte, error) {
|
||
buf := make([]byte, 3+len(p.Payload))
|
||
buf[0] = p.Type
|
||
buf[1] = byte(len(p.Payload) >> 8)
|
||
buf[2] = byte(len(p.Payload))
|
||
copy(buf[3:], p.Payload)
|
||
return buf, nil
|
||
}
|
||
|
||
func main() {
|
||
// 创建服务器配置
|
||
config := network.ServerConfig{
|
||
Network: "tcp",
|
||
Address: ":8080",
|
||
WorkerNum: 100, // 工作协程数量
|
||
QueueSize: 1000, // 任务队列大小
|
||
MaxConn: 10000, // 最大连接数
|
||
ReadTimeout: 30 * time.Second,
|
||
IdleTimeout: 5 * time.Minute,
|
||
}
|
||
|
||
// 创建处理函数
|
||
handler := func(conn network.Conn, packet network.Packet) (network.Packet, error) {
|
||
p := packet.(*MyPacket)
|
||
log.Printf("Received packet: Type=%d, Payload=%s", p.Type, string(p.Payload))
|
||
|
||
// 创建响应
|
||
response := &MyPacket{
|
||
Type: p.Type,
|
||
Payload: []byte("Response: " + string(p.Payload)),
|
||
}
|
||
return response, nil
|
||
}
|
||
|
||
// 创建数据包工厂函数
|
||
packetFactory := func() network.Packet {
|
||
return &MyPacket{}
|
||
}
|
||
|
||
// 创建并启动服务器
|
||
server := network.NewServer(config, handler, packetFactory)
|
||
if err := server.Start(); err != nil {
|
||
log.Fatalf("Failed to start server: %v", err)
|
||
}
|
||
|
||
// 等待信号退出
|
||
// ...
|
||
|
||
// 优雅关闭
|
||
server.Stop()
|
||
}
|
||
```
|
||
|
||
### UDP 服务器示例
|
||
|
||
```go
|
||
package main
|
||
|
||
import (
|
||
"log"
|
||
"time"
|
||
|
||
"github.com/yourusername/network"
|
||
)
|
||
|
||
func main() {
|
||
// 创建服务器配置
|
||
config := network.ServerConfig{
|
||
Network: "udp",
|
||
Address: ":8081",
|
||
WorkerNum: 50,
|
||
QueueSize: 500,
|
||
ReadTimeout: 5 * time.Second,
|
||
}
|
||
|
||
// 使用与 TCP 示例相同的 MyPacket 和 handler
|
||
|
||
// 创建并启动服务器
|
||
server := network.NewServer(config, handler, packetFactory)
|
||
if err := server.Start(); err != nil {
|
||
log.Fatalf("Failed to start UDP server: %v", err)
|
||
}
|
||
|
||
// ...
|
||
}
|
||
```
|
||
|
||
## API 文档
|
||
|
||
### 核心接口
|
||
|
||
#### `Packet` 接口
|
||
|
||
```go
|
||
type Packet interface {
|
||
Decode(data []byte) error
|
||
Encode() ([]byte, error)
|
||
}
|
||
```
|
||
|
||
所有数据包必须实现此接口，用于数据的序列化和反序列化。
|
||
|
||
#### `Conn` 接口
|
||
|
||
```go
|
||
type Conn interface {
|
||
Read(b []byte) (n int, err error)
|
||
Write(b []byte) (n int, err error)
|
||
Close() error
|
||
RemoteAddr() net.Addr
|
||
}
|
||
```
|
||
|
||
表示一个网络连接，TCP 和 UDP 连接都实现了此接口。
|
||
|
||
#### `Handler` 函数类型
|
||
|
||
```go
|
||
type Handler func(conn Conn, packet Packet) (Packet, error)
|
||
```
|
||
|
||
用户定义的处理函数，处理接收到的数据包并返回响应。
|
||
|
||
### 主要结构体
|
||
|
||
#### `ServerConfig` 结构体
|
||
|
||
```go
|
||
type ServerConfig struct {
|
||
Network string // 网络类型: "tcp", "tcp4", "tcp6", "udp", "udp4", "udp6"
|
||
Address string // 监听地址，如 ":8080"
|
||
WorkerNum int // 工作协程数量
|
||
QueueSize int // 任务队列大小
|
||
MaxConn int // 最大连接数
|
||
ReadTimeout time.Duration // 读取超时
|
||
WriteTimeout time.Duration // 写入超时
|
||
IdleTimeout time.Duration // 空闲连接超时
|
||
}
|
||
```
|
||
|
||
#### `HighConcurrentServer` 结构体
|
||
|
||
主要的服务器实现，提供以下方法：
|
||
|
||
- `NewServer(config ServerConfig, handler Handler, packetType func() Packet) *HighConcurrentServer`：创建新服务器
|
||
- `Start() error`：启动服务器
|
||
- `Stop()`：停止服务器
|
||
- `GetActiveConnections() int64`：获取当前活动连接数
|
||
|
||
#### `WorkerPool` 结构体
|
||
|
||
工作协程池，用于高效处理任务：
|
||
|
||
- `NewWorkerPool(size int, queueSize int) *WorkerPool`：创建工作池
|
||
- `Submit(task func())`：提交任务
|
||
- `Stop()`：停止工作池
|
||
|
||
## 性能优化
|
||
|
||
该框架采用了多种技术来优化性能：
|
||
|
||
1. **连接分片**：使用多个分片管理连接，每个分片有自己的锁，减少锁竞争
|
||
2. **工作协程池**：重用协程处理请求，避免频繁创建和销毁协程的开销
|
||
3. **缓冲区池**：使用 `sync.Pool` 复用缓冲区，减少内存分配
|
||
4. **非阻塞 I/O**：使用 channel 进行非阻塞的读写操作
|
||
5. **批量处理**：在可能的情况下批量处理数据包，减少系统调用
|
||
|
||
## 贡献指南
|
||
|
||
欢迎贡献代码、报告问题或提出改进建议。请遵循以下步骤：
|
||
|
||
1. Fork 仓库
|
||
2. 创建功能分支 (`git checkout -b feature/amazing-feature`)
|
||
3. 提交更改 (`git commit -m 'Add some amazing feature'`)
|
||
4. 推送到分支 (`git push origin feature/amazing-feature`)
|
||
5. 创建 Pull Request
|
||
|
||
## 许可证
|
||
|
||
[MIT](LICENSE) |