nets

package module
v1.2.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 29, 2026 License: Apache-2.0 Imports: 23 Imported by: 0

README

NETS

多协议 all-in-one 网络框架
一站式启动 TCP / WebSocket / HTTP / KCP 服务，专注路由、连接与生命周期管理
稳定、快速、安全

License Stars

✨ 特性

特性 说明
多协议统一 TCP、WebSocket、HTTP、KCP 四种协议共用路由与连接管理
路由解耦 通过 msgId 绑定消息工厂与业务处理函数，扩展性强
编解码可选 JSON / Protobuf 一键切换，支持自定义 DataPack
连接管理 分片哈希表存储，读写任务三协程分离，属性存取
限流控制 基于 QPS 的连接限流，支持回调与自动断开
优雅退出 监听系统信号，在进程退出前并行关闭所有服务与连接
高性能 4核8G单台机器 CPS(每秒新建连接数) 可达 10,000+

🔔 环境要求

  • Go ≥ 1.21.1

🚀 快速上手

  1. 安装
go get github.com/451008604/nets
  1. 启动
package main

import (
    "github.com/451008604/nets"
    "github.com/451008604/nets/internal"
    "google.golang.org/protobuf/proto"
)

func main() {
    // 1. 注册路由
    nets.GetInstanceMsgHandler().AddRouter(int32(internal.Test_MsgId_Test_Echo), func() proto.Message { return &internal.Test_EchoRequest{} }, func(conn nets.IConnection, message proto.Message) {
        // 获取请求数据
        msgReq, _ := message.(*internal.Test_EchoRequest)
        // 构造响应数据
        msgRes := &internal.Test_EchoResponse{}
        // 业务处理完毕发送响应数据
        defer conn.SendMsg(int32(internal.Test_MsgId_Test_Echo), msgRes)

        // ...处理逻辑，并设置响应数据...
        msgRes.Message = msgReq.GetMessage()
    })

    // 2. 启动服务(阻塞主协程) 
    nets.GetInstanceServerManager().RegisterServer(nets.GetServerHTTP(), nets.GetServerKCP(), nets.GetServerTCP(), nets.GetServerWS())
}
  1. 详细用法参考

🔧 分布式压测

性能测试工具位于 test 目录，采用 docker compose 编排 1个server + N个client 模拟海量客户端并发测试。具体测试配置项位于 docker-compose.yml
进入 test 目录

  • 执行 sh run_all.sh 启动完整性能测试
  • 执行 sh run_client.sh 单独启动客户端
# 查看内存占用
go tool pprof -http=:8080 http://localhost:16060/debug/pprof/heap

# 查看 server 日志
docker compose logs server -f

📄 许可证

Apache-2.0 License

Documentation

Overview

Package nets provides a lightweight goroutine worker pool for concurrent task execution. It manages a fixed number of worker goroutines that process tasks from buffered channels. Package nets 提供了一个轻量级 goroutine 工作池，用于并发任务执行。 它管理固定数量的工作协程，从缓冲通道中处理任务。

Index

Constants

This section is empty.

Variables

View Source 
var (
	ConnPropertyHttpAuthorization = "HttpAuthorization"
	ConnPropertyHttpReader        = "HttpReader"
	ConnPropertyHttpWriter        = "HttpWriter"
)
View Source 
var (
	ErrPoolClosed = errors.New("worker pool is closed")
)

Common errors returned by WorkerPool methods. WorkerPool 方法返回的常见错误。

Functions

func GenerateConnID added in v1.2.2 

func GenerateConnID() string

func PutMessage added in v1.2.2 

func PutMessage(m IMessage)

func SetCustomServer 

func SetCustomServer(custom *CustomServer)

SetCustomServer applies the caller-provided configuration. IMPORTANT: Must be called BEFORE starting any server. Calling after server start causes data races. Callers should obtain the defaults via GetServerConf() and mutate the fields they need, then pass the result here. This avoids the "zero value means unset" ambiguity, so values like ProtocolIsJson=false or Port=0 are honored as-is.

SetCustomServer 应用调用方提供的配置。 调用方应通过 GetServerConf() 获取默认值并按需修改字段后传入。 这样可避免"零值即未设置"的歧义，使 ProtocolIsJson=false、Port=0 等值被如实采用。 调用方应通过 GetServerConf() 获取默认值并按需修改字段后传入。 这样可避免“零值即未设置”的歧义，使 ProtocolIsJson=false、Port=0 等值被如实采用。

Types

type AppConf 

type AppConf struct {
	AppName          string     // Service Name / 服务名称
	MaxPackSize      uint       // Max Packet Length / 数据包最大长度
	MaxConn          uint       // Max Connections / 最大允许连接数
	WorkerTaskMaxLen uint       // Max Tasks per Worker Queue / 每个工作队列可执行最大任务数量
	WorkerPoolSize   uint       // Worker Pool Size (default CPU*10) / 协程池工作协程数量（默认 CPU*10）
	MaxMsgChanLen    uint       // Max Message Channel Length / 读写通道最大限度
	MaxFlowSecond    int        // Max Requests per Second / 每秒允许的最大请求数量
	ProtocolIsJson   bool       // Use JSON Protocol / 是否使用json协议
	ConnRWTimeOut    uint       // Connection Read/Write Timeout (seconds) / 连接读写超时时间(秒)
	LogLevel         slog.Level // Log Level (default Info) / 日志级别（默认 Info）
	ServerTCP        ServerConf // TCP Service / tcp服务
	ServerWS         ServerConf // WebSocket Service / websocket服务
	ServerHTTP       ServerConf // HTTP Service / http服务
	ServerKCP        ServerConf // KCP Service / kcp服务
}

func GetServerConf 

func GetServerConf() AppConf

Get Default Configuration / 获取默认配置

type BaseRouter 

type BaseRouter struct {
	// contains filtered or unexported fields
}

func (*BaseRouter) GetNewMsg 

func (b *BaseRouter) GetNewMsg() proto.Message

func (*BaseRouter) RunHandler 

func (b *BaseRouter) RunHandler(conn IConnection, message proto.Message)

func (*BaseRouter) SetHandler 

func (b *BaseRouter) SetHandler(msgHandler IReceiveMsgHandler)

func (*BaseRouter) SetMsg 

func (b *BaseRouter) SetMsg(msgTemplate INewMsgStructTemplate)

type ConnectionBase 

type ConnectionBase struct {
	// contains filtered or unexported fields
}

func (*ConnectionBase) ByteToProtocol 

func (c *ConnectionBase) ByteToProtocol(byte []byte, target proto.Message) error

func (*ConnectionBase) Close added in v1.2.2 

func (c *ConnectionBase) Close()

func (*ConnectionBase) ConnCtx added in v1.2.4 

func (c *ConnectionBase) ConnCtx() context.Context

func (*ConnectionBase) DoTask 

func (c *ConnectionBase) DoTask(task func()) bool

func (*ConnectionBase) FlowControl 

func (c *ConnectionBase) FlowControl() bool

func (*ConnectionBase) GetConnId 

func (c *ConnectionBase) GetConnId() string

func (*ConnectionBase) GetProperty 

func (c *ConnectionBase) GetProperty(key string) any

func (*ConnectionBase) IsClose 

func (c *ConnectionBase) IsClose() bool

func (*ConnectionBase) Open added in v1.2.2 

func (c *ConnectionBase) Open()

func (*ConnectionBase) ProtocolToByte 

func (c *ConnectionBase) ProtocolToByte(str proto.Message) []byte

func (*ConnectionBase) RemoteAddrStr added in v1.2.2 

func (c *ConnectionBase) RemoteAddrStr() string

func (*ConnectionBase) RemoveProperty added in v1.2.2 

func (c *ConnectionBase) RemoveProperty(key string)

func (*ConnectionBase) SendMsg 

func (c *ConnectionBase) SendMsg(msgId int32, msgData proto.Message)

func (*ConnectionBase) SetProperty added in v1.2.2 

func (c *ConnectionBase) SetProperty(key string, value any)

type ConnectionManager 

type ConnectionManager struct {
	// contains filtered or unexported fields
}

func GetInstanceConnManager 

func GetInstanceConnManager() *ConnectionManager

Connection Manager / 连接管理器

func (*ConnectionManager) Add 

func (c *ConnectionManager) Add(conn IConnection)

func (*ConnectionManager) ClearConn 

func (c *ConnectionManager) ClearConn()

func (*ConnectionManager) ConnRateLimiting 

func (c *ConnectionManager) ConnRateLimiting(conn IConnection)

func (*ConnectionManager) Get 

func (c *ConnectionManager) Get(connId string) (IConnection, bool)

func (*ConnectionManager) GetConnClosed added in v1.2.2 

func (c *ConnectionManager) GetConnClosed(conn IConnection)

func (*ConnectionManager) GetConnOpened added in v1.2.2 

func (c *ConnectionManager) GetConnOpened(conn IConnection)

func (*ConnectionManager) Len 

func (c *ConnectionManager) Len() int

func (*ConnectionManager) RangeConnections 

func (c *ConnectionManager) RangeConnections(handler func(conn IConnection))

func (*ConnectionManager) Remove 

func (c *ConnectionManager) Remove(conn IConnection)

func (*ConnectionManager) SetConnClosed added in v1.2.2 

func (c *ConnectionManager) SetConnClosed(connCloseCallBack func(conn IConnection))

func (*ConnectionManager) SetConnOnRateLimiting 

func (c *ConnectionManager) SetConnOnRateLimiting(limitCallBack func(conn IConnection))

func (*ConnectionManager) SetConnOpened added in v1.2.2 

func (c *ConnectionManager) SetConnOpened(connOpenCallBack func(conn IConnection))

type CustomServer 

type CustomServer struct {
	AppConf  AppConf   // Service Startup Configuration / 服务启动配置
	DataPack IDataPack // Custom Encoder/Decoder / 自定义编码/解码器
}

Custom Server / 自定义服务器

type IConnection 

type IConnection interface {
	// Start Connection (called by connmanager) / 启动连接(通过connmanager调用)
	Open()
	// Stop Connection (called by connmanager) / 停止连接(通过connmanager调用)
	Close()

	// Get Real Connection / 获取真实连接
	GetNetConn() net.Conn
	// Connection Context / 连接上下文
	ConnCtx() context.Context

	// Start Message Receiving Goroutine / 启动接收消息协程
	StartReader() bool
	// Start Message Sending Goroutine / 启动发送消息协程
	StartWriter(data []byte) bool
	// Execute Task / 执行任务
	DoTask(task func()) bool

	// Get Current Connection ID / 获取当前连接Id
	GetConnId() string
	// Get Client Address Info / 获取客户端地址信息
	RemoteAddrStr() string
	// Get Whether Connection is Closed / 获取连接是否已关闭
	IsClose() bool
	// Get Connection Bound Property / 获取连接绑定的属性
	GetProperty(key string) any
	// Set Connection Bound Property / 设置连接绑定的属性
	SetProperty(key string, value any)
	// Remove Connection Bound Property / 移除连接绑定的属性
	RemoveProperty(key string)

	// Send Message to Client / 发送消息给客户端
	SendMsg(msgId int32, msgData proto.Message)

	// Rate Limiting Control / 限流控制
	FlowControl() bool

	// Serialize / 序列化
	ProtocolToByte(str proto.Message) []byte
	// Deserialize / 反序列化
	ByteToProtocol(byte []byte, target proto.Message) error
}

func NewConnectionHTTP 

func NewConnectionHTTP(server IServer, writer http.ResponseWriter, reader *http.Request) IConnection

func NewConnectionKCP 

func NewConnectionKCP(server *serverKCP, conn net.Conn) IConnection

func NewConnectionTCP 

func NewConnectionTCP(server IServer, conn *net.TCPConn) IConnection

func NewConnectionWS 

func NewConnectionWS(server IServer, conn *websocket.Conn) IConnection

type IDataPack 

type IDataPack interface {
	// Get Message Header Length / 获取消息头长度
	GetHeadLen() int
	// Message Pack / 消息封包
	Pack(msg IMessage) []byte
	// Message Unpack / 消息拆包
	UnPack([]byte) IMessage
}

Pack/Unpack, obtain message data through fixed packet header to solve TCP sticky packet problem / 封包拆包，通过固定的包头获取消息数据，解决TCP粘包问题

func NewDataPack 

func NewDataPack() IDataPack

type IErrCapture 

type IErrCapture func(conn IConnection, recover any)

type IFilter 

type IFilter func(conn IConnection, msg IMessage) bool

type IMessage 

type IMessage interface {
	// Get Message ID / 获取消息Id
	GetMsgId() uint16
	// Get Message Length / 获取消息长度
	GetDataLen() uint16
	// Get Message Content / 获取消息内容
	GetData() []byte
	// Set Message Content / 设置消息内容
	SetData([]byte)
}

Define Message Template / 定义消息模板

type INewMsgStructTemplate 

type INewMsgStructTemplate func() proto.Message

type IReceiveMsgHandler 

type IReceiveMsgHandler func(conn IConnection, message proto.Message)

type IServer 

type IServer interface {
	// Get Server Name / 获取服务器名称
	GetServerName() string
	// Start Server / 启动服务器
	Start()
}

Define Server Interface / 定义服务器接口

func GetServerHTTP 

func GetServerHTTP() IServer

func GetServerKCP 

func GetServerKCP() IServer

func GetServerTCP 

func GetServerTCP() IServer

func GetServerWS 

func GetServerWS() IServer

type Message 

type Message struct {
	proto.Message `json:"-"`
	Id            uint16 `protobuf:"bytes,1,opt,name=msg_id,proto3" json:"msg_id"` // Message ID / 消息Id
	Data          []byte `protobuf:"bytes,2,opt,name=data,proto3" json:"data"`     // Message Content / 消息内容
	DataLen       uint16 `json:"-"`                                                // Message Length / 消息长度
}

func GetMessage added in v1.2.2 

func GetMessage() *Message

func (*Message) GetData 

func (m *Message) GetData() []byte

func (*Message) GetDataLen 

func (m *Message) GetDataLen() uint16

func (*Message) GetMsgId 

func (m *Message) GetMsgId() uint16

func (*Message) SetData 

func (m *Message) SetData(bytes []byte)

type MsgHandler 

type MsgHandler struct {
	// contains filtered or unexported fields
}

func GetInstanceMsgHandler 

func GetInstanceMsgHandler() *MsgHandler

Message Handler / 消息处理器

func (*MsgHandler) AddRouter 

func (m *MsgHandler) AddRouter(msgId int32, msgTemplate INewMsgStructTemplate, msgHandler IReceiveMsgHandler)

func (*MsgHandler) GetErrCapture 

func (m *MsgHandler) GetErrCapture(conn IConnection)

func (*MsgHandler) GetFilter 

func (m *MsgHandler) GetFilter(conn IConnection, msg IMessage) bool

func (*MsgHandler) GetRouter added in v1.2.4 

func (m *MsgHandler) GetRouter(msgId int32) (*BaseRouter, bool)

func (*MsgHandler) SetErrCapture 

func (m *MsgHandler) SetErrCapture(fun IErrCapture)

func (*MsgHandler) SetFilter 

func (m *MsgHandler) SetFilter(fun IFilter)

type ServerConf 

type ServerConf struct {
	Address     string // IP Address / IP地址
	Port        int    // Port / 端口
	TLSCertPath string // SSL Certificate Path / ssl证书路径
	TLSKeyPath  string // SSL Key Path / ssl密钥路径
}

type ServerManager 

type ServerManager struct {
	// contains filtered or unexported fields
}

func GetInstanceServerManager 

func GetInstanceServerManager() *ServerManager

Server Manager / 服务管理器

func (*ServerManager) IsClose 

func (c *ServerManager) IsClose() bool

func (*ServerManager) RegisterServer 

func (c *ServerManager) RegisterServer(server ...IServer)

func (*ServerManager) StopAll 

func (c *ServerManager) StopAll()

func (*ServerManager) WaitGroupAdd 

func (c *ServerManager) WaitGroupAdd(delta int)

func (*ServerManager) WaitGroupDone 

func (c *ServerManager) WaitGroupDone()

type WorkerPool added in v1.2.4 

type WorkerPool struct {
	// contains filtered or unexported fields
}

WorkerPool manages a pool of goroutine workers that execute submitted tasks concurrently. It uses done channel pattern to prevent send-on-closed-channel panics.

Features:

  • Submit(): Round-robin distribution for general tasks
  • SubmitWithWorker(): Hash binding for ordered execution (same workerId → same worker)
  • SubmitCtx(): Context-aware submission with cancellation support
  • HashWorkerId(): Utility to convert string to valid workerId

WorkerPool 管理一组 goroutine 工作协程，并发执行提交的任务。 使用 done channel 模式防止 send-on-closed-channel panic。

特性：

  • Submit(): 轮询分配，适用于普通任务
  • SubmitWithWorker(): 哈希绑定，保证顺序执行（相同 workerId → 相同 worker）
  • SubmitCtx(): 支持 context 取消的提交
  • HashWorkerId(): 将字符串转换为有效 workerId 的工具方法

func GetInstanceWorkerPool added in v1.2.4 

func GetInstanceWorkerPool() *WorkerPool

GetInstanceWorkerPool returns the singleton instance of WorkerPool. It initializes the pool on first call using configuration from defaultServer. If WorkerPoolSize is not configured (<= 0), it defaults to runtime.NumCPU() * 10.

GetInstanceWorkerPool 返回 WorkerPool 的单例实例。 它在第一次调用时使用 defaultServer 的配置初始化池。 如果 WorkerPoolSize 未配置（<= 0），则默认为 runtime.NumCPU() * 10。

func NewWorkerPool added in v1.2.4 

func NewWorkerPool(poolSize, maxTaskLen uint) *WorkerPool

NewWorkerPool creates a new WorkerPool with the specified number of workers and task channel capacity. It immediately starts all worker goroutines which will process tasks submitted to the pool.

NewWorkerPool 使用指定的工作协程数量和任务通道容量创建一个新的 WorkerPool。 它立即启动所有工作协程，这些协程将处理提交到池中的任务。

func (*WorkerPool) HashWorkerId added in v1.2.4 

func (p *WorkerPool) HashWorkerId(key string) int

HashWorkerId converts a string identifier to an integer. Use this to generate workerId from connection ID or other string keys. The returned value can be used with SubmitWithWorker, which will map it to a valid worker internally.

HashWorkerId 使用 FNV 哈希将字符串标识符转换为整数。 用于从连接 ID 或其他字符串键生成 workerId。 返回值可用于 SubmitWithWorker，后者会在内部将其映射到有效的 worker。

func (*WorkerPool) Pending added in v1.2.4 

func (p *WorkerPool) Pending() int

Pending returns the total number of tasks waiting in all worker queues.

Pending 返回所有 worker 队列中等待的任务总数。

func (*WorkerPool) Stats added in v1.2.4 

func (p *WorkerPool) Stats() WorkerPoolStats

Stats returns a snapshot of the pool's current statistics.

Stats 返回池当前统计信息的快照。

func (*WorkerPool) Stop added in v1.2.4 

func (p *WorkerPool) Stop()

Stop gracefully shuts down the worker pool. It prevents new task submissions, waits for all workers to finish their current tasks, and then returns. Subsequent calls to Stop are no-ops.

Stop 优雅地关闭工作池。它阻止新任务提交，等待所有工作协程完成当前任务，然后返回。 后续对 Stop 的调用是无操作。

func (*WorkerPool) Submit added in v1.2.4 

func (p *WorkerPool) Submit(task func()) error

Submit adds a task to the pool for execution using round-robin distribution. It blocks if the target worker's channel is full. Returns ErrPoolClosed if the pool has been closed. Priority: p.done > channel send

Submit 使用轮询分配将任务添加到池中执行。 如果目标 worker 的通道已满则阻塞。 如果池已关闭则返回 ErrPoolClosed。 优先级：p.done > channel send

func (*WorkerPool) SubmitCtx added in v1.2.4 

func (p *WorkerPool) SubmitCtx(ctx context.Context, task func()) error

SubmitCtx adds a task to the pool with context cancellation support using round-robin distribution. It blocks if the target worker's channel is full, but can be canceled via the context. Priority: ctx.Done() > p.done > channel send

SubmitCtx 使用轮询分配将任务添加到池中，支持 context 取消。 如果目标 worker 的通道已满则阻塞，但可以通过 context 取消。 优先级：ctx.Done() > p.done > channel send

func (*WorkerPool) SubmitWithWorker added in v1.2.4 

func (p *WorkerPool) SubmitWithWorker(task func(), workerId int) error

SubmitWithWorker adds a task bound to a specific worker. Tasks with the same workerId are guaranteed to execute on the same worker in FIFO order. The workerId will be mapped to a valid worker index internally using modulo operation. Priority: p.done > channel send

Use HashWorkerId() to convert string (e.g., connection ID) to workerId.

SubmitWithWorker 添加绑定到特定 worker 的任务。 相同 workerId 的任务保证在同一 worker 上按 FIFO 顺序执行。 workerId 会在内部通过取模运算映射到有效的 worker 索引。 优先级：p.done > channel send

使用 HashWorkerId() 将字符串（如连接 ID）转换为 workerId。

func (*WorkerPool) SubmitWithWorkerCtx added in v1.2.4 

func (p *WorkerPool) SubmitWithWorkerCtx(ctx context.Context, task func(), workerId int) error

SubmitWithWorkerCtx adds a task bound to a specific worker with context cancellation support. Priority: ctx.Done() > p.done > channel send

SubmitWithWorkerCtx 添加绑定到特定 worker 的任务，支持 context 取消。 优先级：ctx.Done() > p.done > channel send

func (*WorkerPool) TrySubmit added in v1.2.4 

func (p *WorkerPool) TrySubmit(task func()) bool

TrySubmit attempts to add a task without blocking using round-robin distribution. Returns true if the task was submitted successfully, false if the channel is full or the pool is closed. Priority: p.done > channel send > default

TrySubmit 尝试使用轮询分配非阻塞地添加任务。 如果任务提交成功则返回 true，如果通道已满或池已关闭则返回 false。 优先级：p.done > channel send > default

func (*WorkerPool) TrySubmitWithWorker added in v1.2.4 

func (p *WorkerPool) TrySubmitWithWorker(task func(), workerId int) bool

TrySubmitWithWorker attempts to add a task bound to a specific worker without blocking. The workerId will be mapped to a valid worker index internally using modulo operation. Priority: p.done > channel send > default

TrySubmitWithWorker 尝试非阻塞地添加绑定到特定 worker 的任务。 workerId 会在内部通过取模运算映射到有效的 worker 索引。 优先级：p.done > channel send > default

type WorkerPoolStats added in v1.2.4 

type WorkerPoolStats struct {
	ActiveWorkers  int32
	PendingTasks   int
	Capacity       int32
	TotalSubmitted int64
}

WorkerPoolStats contains statistics about the worker pool's current state. WorkerPoolStats 包含工作池当前状态的统计信息。

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.