Documentation ¶
Index ¶
- Constants
- Variables
- func CreateMessageByID(hash uint32) proto.Message
- func CreateMessageByName(name string) proto.Message
- func CreatePairingAck(req proto.Message) proto.Message
- func CreatePairingAckBy(reqName string) proto.Message
- func DateTime() string
- func DeregisterHandler(command int32)
- func DispatchMessage(ctx context.Context, msg *Packet) error
- func FreePacket(pkt *Packet)
- func GetEnv(key string, def string) string
- func GetEnvBool(key string) bool
- func GetEnvFloat(key string, def float64) float64
- func GetEnvInt(key string, def int) int
- func GetEnvInt64(key string, def int64) int64
- func GetMessageIDByName(name string) uint32
- func GetMessageIDOf(msg proto.Message) uint32
- func GetMessageNameByID(hash uint32) string
- func GetPairingAckName(reqName string) string
- func GetServiceTypes() []uint8
- func IBody2Bytes(body interface{}) []byte
- func IBody2Float(body interface{}) float64
- func IBody2Int64(body interface{}) int64
- func IBody2String(body interface{}) string
- func IBodyAsBytes(body interface{}) []byte
- func Now() time.Time
- func NowString() string
- func PB2JSON(msg proto.Message) string
- func RegisterAllMessages(validNameSuffix ...string) (er error)
- func RegisterHandler(command int32, action PacketHandler)
- func RegisterService(s Service)
- func SafeToBody(val interface{}) interface{}
- func SetPacketAllocPolicy(policy int)
- func SetPanicHandler(f PanicHandler)
- func StartClock()
- func StopClock()
- func WallClock() *datetime.Clock
- type ArenaAllocator
- type Endpoint
- type EndpointFlag
- type EndpointMap
- type IPacketAllocator
- type MessageDispatcher
- func (d *MessageDispatcher) Deregister(command int32)
- func (d *MessageDispatcher) DeregisterOne(command int32, handler PacketHandler)
- func (d *MessageDispatcher) Dispatch(ctx context.Context, pkt *Packet) error
- func (d *MessageDispatcher) Register(command int32, handler PacketHandler)
- func (d *MessageDispatcher) SetPanicHandler(f PanicHandler)
- type MessageEndpoint
- type NetPacket
- type NodeID
- type NodeIDSet
- type Packet
- func (m *Packet) AddRefers(refers ...NodeID)
- func (m *Packet) AutoDecode() error
- func (m *Packet) Clone() *Packet
- func (m *Packet) Decode(msg proto.Message) error
- func (m *Packet) DecodeTo(msg proto.Message) error
- func (m *Packet) EncodeToBytes() []byte
- func (m *Packet) Errno() int32
- func (m *Packet) Refer() NodeID
- func (m *Packet) Refuse(errno int32) error
- func (m *Packet) Reply(ack proto.Message) error
- func (m *Packet) ReplyAny(body interface{}) error
- func (m *Packet) Reset()
- func (m *Packet) SetBody(val interface{})
- func (m *Packet) SetErrno(ec int32)
- func (m *Packet) SetRefer(node NodeID)
- func (m *Packet) String() string
- type PacketFlag
- type PacketHandler
- type PanicHandler
- type PoolAllocator
- type Runnable
- type RuntimeAllocator
- type Service
- type ServiceContext
- func (c *ServiceContext) AddFinalizer(action func())
- func (c *ServiceContext) Close()
- func (c *ServiceContext) EnqueueQueue(pkt *Packet)
- func (c *ServiceContext) EnqueueTask(task Runnable)
- func (c *ServiceContext) InboundQueue() chan<- *Packet
- func (c *ServiceContext) Instance() Service
- func (c *ServiceContext) MessageQueue() <-chan *Packet
- func (c *ServiceContext) OutboundQueue() <-chan *Packet
- func (c *ServiceContext) QuitDone() <-chan struct{}
- func (c *ServiceContext) RunID() string
- func (c *ServiceContext) SendPacket(pkt *Packet)
- func (c *ServiceContext) SetInstance(inst Service)
- func (c *ServiceContext) StartTime() time.Time
- func (c *ServiceContext) TaskQueue() <-chan Runnable
- type State
- type Task
- type Timer
- type TimerQueue
- func (s *TimerQueue) Cancel(id int) bool
- func (s *TimerQueue) Chan() <-chan Runnable
- func (s *TimerQueue) IsPending(id int) bool
- func (s *TimerQueue) RunAfter(timeUnits int, r Runnable) int
- func (s *TimerQueue) RunEvery(interval int, r Runnable) int
- func (s *TimerQueue) Shutdown()
- func (s *TimerQueue) Size() int
- func (s *TimerQueue) Start()
Constants ¶
const ( FOREACH_CONTINUE = 0 FOREACH_EXIT = 1 )
foreach控制
const ( SERVICE_ALL = 0xFF // 所有服务 INSTANCE_ALL = 0xFFFF // 所有实例 )
const ( NodeServiceShift = 16 NodeTypeShift = 31 NodeServiceMask = 0x00FF0000 NodeInstanceMask = 0x0000FFFF )
const ( AllocPolicyRuntime = 0 AllocPolicyPool = 1 AllocPolicyArena = 2 )
不同的分配算法(策略)
const ( SuffixREQ = "Req" SuffixACK = "Ack" )
消息协议规则:
1, 请求消息以REQ结尾 2, 响应消息以ACK结尾 3, 通知消息以NTF结尾
const ( StateInit = 0 StateStarting = 1 StateRunning = 2 StateShuttingDown = 3 StateTerminated = 4 )
const ( TaskStateInit = 0 TaskScheduled = 1 // task is scheduled for execution TaskExecuted = 2 // a non-repeating task has already executed (or is currently executing) and has not been cancelled. TaskCancelled = 3 // task has been cancelled (with a call to TimerTask.Cancel). )
const ( PendingQueueCapacity = 128 // pending add/delete TimeoutQueueCapacity = 1024 // pending timed-out )
const ClockEpoch int64 = 1577836800 // 2020-01-01 00:00:00 UTC
epoch of clock
const VERSION = "0.6.3"
版本号
Variables ¶
var ErrContextInboundQueueFull = errors.New("context inbound queue full")
Functions ¶
func CreateMessageByName ¶ added in v0.4.4
根据名称创建消息
func CreatePairingAck ¶ added in v0.4.4
如果消息的名字是XXXReq,则尝试创建与其名称对应的XXXAck消息
func CreatePairingAckBy ¶ added in v0.4.4
如果消息的名字是XXXReq,则尝试创建与其名称对应的XXXAck消息
func DeregisterHandler ¶ added in v0.4.8
func DeregisterHandler(command int32)
func GetEnvInt64 ¶ added in v0.4.7
func GetPairingAckName ¶ added in v0.4.4
根据Req消息的名称,返回其对应的Ack消息名称
func IBodyAsBytes ¶ added in v0.4.1
func IBodyAsBytes(body interface{}) []byte
func RegisterAllMessages ¶ added in v0.6.1
自动注册所有protobuf消息 因为protobuf使用init()注册(RegisterType),此API需要在import后调用
func RegisterHandler ¶ added in v0.4.8
func RegisterHandler(command int32, action PacketHandler)
func SetPanicHandler ¶ added in v0.4.8
func SetPanicHandler(f PanicHandler)
Types ¶
type ArenaAllocator ¶ added in v0.4.2
type ArenaAllocator struct {
// contains filtered or unexported fields
}
arena算法的分配器 思路是:
一次申请一个block(N个元素的数组),然后从block数组里再逐个按需分配,block分配完了就丢掉(交给GC),再申请另一个block 这样对runtime来说每次malloc都是以N个元素大小的单位,理论上可以减缓GC的压力,并且业务层无需做Free
func NewArenaAllocator ¶ added in v0.4.2
func NewArenaAllocator(blockSize int) *ArenaAllocator
type Endpoint ¶
type Endpoint interface { MessageEndpoint // 原始连接对象 RawConn() net.Conn // 发送/接收计数数据 Stats() *stats.Stats ErrorChan() <-chan error OutMsgChan() chan<- *Packet // 开启read/write线程 Go(EndpointFlag) // 设置加解密 SetEncryptPair(cipher.BlockCryptor, cipher.BlockCryptor) }
网络端点
type EndpointFlag ¶
type EndpointFlag uint32
开启reader/writer标记
const ( EndpointReader EndpointFlag = 0x01 // 只开启reader EndpointWriter EndpointFlag = 0x02 // 只开启writer EndpointReadWriter EndpointFlag = 0x03 // 开启reader和writer )
type EndpointMap ¶
type EndpointMap interface { Size() int Has(node NodeID) bool Get(node NodeID) Endpoint IsEmpty() bool Foreach(func(Endpoint) int) Put(node NodeID, endpoint Endpoint) PutIfAbsent(node NodeID, endpoint Endpoint) Remove(node NodeID) Pop(node NodeID) Endpoint Clear() }
线程安全的Endpoint字典
type IPacketAllocator ¶ added in v0.6.1
func GetPacketAllocator ¶ added in v0.6.1
func GetPacketAllocator(policy int) IPacketAllocator
type MessageDispatcher ¶ added in v0.4.8
type MessageDispatcher struct {
// contains filtered or unexported fields
}
消息派发
func DefaultDispatcher ¶ added in v0.4.8
func DefaultDispatcher() *MessageDispatcher
func NewMessageDispatcher ¶ added in v0.4.8
func NewMessageDispatcher() *MessageDispatcher
func (*MessageDispatcher) Deregister ¶ added in v0.4.8
func (d *MessageDispatcher) Deregister(command int32)
取消所有
func (*MessageDispatcher) DeregisterOne ¶ added in v0.4.8
func (d *MessageDispatcher) DeregisterOne(command int32, handler PacketHandler)
取消单个注册
func (*MessageDispatcher) Dispatch ¶ added in v0.4.8
func (d *MessageDispatcher) Dispatch(ctx context.Context, pkt *Packet) error
func (*MessageDispatcher) Register ¶ added in v0.4.8
func (d *MessageDispatcher) Register(command int32, handler PacketHandler)
注册一个
func (*MessageDispatcher) SetPanicHandler ¶ added in v0.4.8
func (d *MessageDispatcher) SetPanicHandler(f PanicHandler)
type MessageEndpoint ¶
type MessageEndpoint interface { // 节点号 Node() NodeID SetNode(NodeID) // 远端地址 RemoteAddr() string // 发送消息(非阻塞) Send(*Packet) error // 关闭读/写 Close() error ForceClose(error) IsRunning() bool // 绑定自定义数据 SetUserData(interface{}) UserData() interface{} }
绑定到消息上的endpoint
type NodeID ¶
type NodeID uint32
节点ID 一个32位整数表示的节点号,用以标识一个service(最高位为0),或者一个客户端session(最高位为1) 如果是服务编号:8位服务编号,16位服务实例编号
服务实例二进制布局 -------------------------------------- | reserved | service | instance | -------------------------------------- 32 24 16 0
type Packet ¶ added in v0.6.1
type Packet struct { Command int32 `json:"cmd"` // 协议命令,即如何执行消息 MsgID uint32 `json:"mid,omitempty"` // 协议ID,即如何解析body Seq uint16 `json:"seq,omitempty"` // 会话内的唯一序列号 Flags PacketFlag `json:"flg,omitempty"` // 标志位 Node NodeID `json:"node,omitempty"` // 源/目标节点 Refers []NodeID `json:"ref,omitempty"` // 组播session列表 Body interface{} `json:"body,omitempty"` // 消息内容,int32/int64/float64/string/bytes/proto.Packet Endpoint MessageEndpoint `json:"-"` // 关联的endpoint }
Packet表示一个应用层消息
func NewPacket ¶ added in v0.6.1
func NewPacket(cmd int32, flags PacketFlag, seq uint16, body interface{}) *Packet
func (*Packet) AutoDecode ¶ added in v0.6.1
根据MsgID把body的字节流解析为proto消息
func (*Packet) EncodeToBytes ¶ added in v0.6.1
type PacketFlag ¶
type PacketFlag uint16
消息标志位
const ( // 低8位用于表达一些传输flag PFlagCompressed PacketFlag = 0x01 // 压缩 PFlagEncrypted PacketFlag = 0x02 // 加密 PFlagError PacketFlag = 0x04 // 错误标记 PFlagHashCmd PacketFlag = 0x10 // CMD是字符串hash PFlagReserve PacketFlag = 0x20 // // 高8位用于server之间通信使用 PFlagRPC PacketFlag = 0x0100 // 远程调用 PFlagMulticast PacketFlag = 0x1000 // 组播消息 PFlagRoute PacketFlag = 0x2000 // 路由消息 )
func (PacketFlag) Clear ¶ added in v0.1.21
func (g PacketFlag) Clear(n PacketFlag) PacketFlag
func (PacketFlag) Has ¶ added in v0.1.21
func (g PacketFlag) Has(n PacketFlag) bool
type PanicHandler ¶ added in v0.4.8
type PanicHandler func(*Packet, interface{})
type PoolAllocator ¶ added in v0.4.2
type PoolAllocator struct {
// contains filtered or unexported fields
}
使用sync.Pool的分配器
func NewPoolAllocator ¶ added in v0.4.2
func NewPoolAllocator() *PoolAllocator
func (*PoolAllocator) Alloc ¶ added in v0.4.2
func (a *PoolAllocator) Alloc() *Packet
func (*PoolAllocator) Free ¶ added in v0.4.2
func (a *PoolAllocator) Free(msg *Packet)
放回pool里供后面的Alloc使用
type RuntimeAllocator ¶ added in v0.4.2
type RuntimeAllocator struct {
// contains filtered or unexported fields
}
使用new函数的分配器
func NewRuntimeAllocator ¶ added in v0.4.2
func NewRuntimeAllocator() *RuntimeAllocator
func (*RuntimeAllocator) Alloc ¶ added in v0.4.2
func (a *RuntimeAllocator) Alloc() *Packet
func (*RuntimeAllocator) Free ¶ added in v0.4.2
func (a *RuntimeAllocator) Free(msg *Packet)
type Service ¶
type Service interface { Type() uint8 // 节点号 Node() NodeID SetNode(id NodeID) // 当前状态 State() int32 SetState(int32) // 上下文对象 Context() *ServiceContext // 初始化 Init(context.Context, *ServiceContext) error // 启动服务 Startup(context.Context) error }
Service实现具体的服务
type ServiceContext ¶
type ServiceContext struct {
// contains filtered or unexported fields
}
服务的上下文
func NewServiceContext ¶
func NewServiceContext(msgQueueSize, taskQueueSize int) *ServiceContext
func (*ServiceContext) AddFinalizer ¶ added in v0.1.28
func (c *ServiceContext) AddFinalizer(action func())
func (*ServiceContext) EnqueueQueue ¶ added in v0.6.2
func (c *ServiceContext) EnqueueQueue(pkt *Packet)
投递一条待处理的消息
func (*ServiceContext) EnqueueTask ¶ added in v0.6.2
func (c *ServiceContext) EnqueueTask(task Runnable)
func (*ServiceContext) InboundQueue ¶ added in v0.1.6
func (c *ServiceContext) InboundQueue() chan<- *Packet
用于在select语句里投入待处理队列
func (*ServiceContext) Instance ¶ added in v0.1.2
func (c *ServiceContext) Instance() Service
service实例
func (*ServiceContext) MessageQueue ¶
func (c *ServiceContext) MessageQueue() <-chan *Packet
消息队列,仅消费
func (*ServiceContext) OutboundQueue ¶ added in v0.6.2
func (c *ServiceContext) OutboundQueue() <-chan *Packet
待发送队列
func (*ServiceContext) QuitDone ¶ added in v0.1.24
func (c *ServiceContext) QuitDone() <-chan struct{}
等待close完成
func (*ServiceContext) SendPacket ¶ added in v0.6.1
func (c *ServiceContext) SendPacket(pkt *Packet)
投递一条待发送的消息
func (*ServiceContext) SetInstance ¶ added in v0.2.9
func (c *ServiceContext) SetInstance(inst Service)
func (*ServiceContext) StartTime ¶ added in v0.1.24
func (c *ServiceContext) StartTime() time.Time
func (*ServiceContext) TaskQueue ¶ added in v0.6.2
func (c *ServiceContext) TaskQueue() <-chan Runnable
任务队列,仅消费
type State ¶ added in v0.1.19
type State int32
service state
func (*State) IsShuttingDown ¶ added in v0.1.19
func (*State) IsTerminated ¶ added in v0.1.19
type Timer ¶ added in v0.4.8
type Timer interface { Start() // 关闭定时器 Shutdown() // 在`timeUnits`时间后执行`r` RunAfter(timeUnits int, r Runnable) int // 每隔`interval`时间执行`r` RunEvery(interval int, r Runnable) int // 取消一个timer Cancel(id int) bool // 判断timer是否在计划中 IsPending(id int) bool // 超时的待执行runner Chan() <-chan Runnable // timer数量 Size() int }
定时器
func NewDefaultTimerQueue ¶ added in v0.4.8
func NewDefaultTimerQueue() Timer
func NewTimerQueue ¶ added in v0.4.8
type TimerQueue ¶ added in v0.4.8
type TimerQueue struct { C chan Runnable // 到期的定时器 // contains filtered or unexported fields }
最小堆实现的定时器 标准库的四叉堆实现的time.Timer已经可以满足大多数高精度的定时需求 这个实现主要是为了在大量timer的场景,把timer的压力从runtime放到应用上
func (*TimerQueue) Chan ¶ added in v0.4.8
func (s *TimerQueue) Chan() <-chan Runnable
func (*TimerQueue) IsPending ¶ added in v0.4.8
func (s *TimerQueue) IsPending(id int) bool
func (*TimerQueue) RunAfter ¶ added in v0.4.8
func (s *TimerQueue) RunAfter(timeUnits int, r Runnable) int
创建一个定时器,在`timeUnits`时间后运行`r`
func (*TimerQueue) RunEvery ¶ added in v0.4.8
func (s *TimerQueue) RunEvery(interval int, r Runnable) int
创建一个定时器,每隔`interval`时间运行一次`r`
func (*TimerQueue) Shutdown ¶ added in v0.4.8
func (s *TimerQueue) Shutdown()
func (*TimerQueue) Size ¶ added in v0.4.8
func (s *TimerQueue) Size() int
func (*TimerQueue) Start ¶ added in v0.4.8
func (s *TimerQueue) Start()
Starts the background thread explicitly