cellnet

README

cellnet

简单,方便,高效的跨平台服务器网络库

特性

异步单线程多进程架构

• 无需处理繁琐的多线程安全问题

• 底层IO仍然使用goroutine进行处理, 保证IO吞吐率

• 发送时自动合并封包(性能效果决定于实际请求和发送比例)

数据协议

• 封包类型采用Type-Length-Value的私有tcp封包, 自带序列号防御简单的封包复制

• 消息统一使用Protobuf格式进行通信

• 自动生成消息ID

RPC

• 异步远程过程调用

日志

• 分级日志

• 可以方便的通过日志查看收发消息(Protobuf)的每一个字段消息

第三方库依赖

• github.com/golang/protobuf/proto

• github.com/davyxu/golog

• gopkg.in/mgo.v2

性能测试

命令行: go test -v github.com/davyxu/cellnet/benchmark/io

CPU: i7 6700 3.4GHz 8核

测试用例: localhost 1000连接 同时对服务器进行实时PingPong测试

平台: Windows 7 x64/CentOS 6.5 x64

QPS: 13.7w

例子

Echo

func server() {

	pipe := cellnet.NewEventPipe()

	evq := socket.NewAcceptor(pipe).Start("127.0.0.1:7234")

	socket.RegisterSessionMessage(evq, "gamedef.TestEchoACK", func(content interface{}, ses cellnet.Session) {
		msg := content.(*gamedef.TestEchoACK)

		log.Debugln("server recv:", msg.String())

		ses.Send(&gamedef.TestEchoACK{
			Content: msg.String,
		})

	})

	pipe.Start()

}

func client() {

	pipe := cellnet.NewEventPipe()

	evq := socket.NewConnector(pipe).Start("127.0.0.1:7234")

	socket.RegisterSessionMessage(evq, "gamedef.TestEchoACK", func(content interface{}, ses cellnet.Session) {
		msg := content.(*gamedef.TestEchoACK)

		log.Debugln("client recv:", msg.String())

	})

	socket.RegisterSessionMessage(evq, "gamedef.SessionConnected", func(content interface{}, ses cellnet.Session) {

		ses.Send(&gamedef.TestEchoACK{
			Content: "hello",
		})

	})

	pipe.Start()
}

文件夹功能

benchmark\ 性能测试用例

db\ db封装

proto\ cellnet内部的proto

protoc-gen-msg\ 消息id生成

rpc\ 异步远程过程调用封装

socket\ 套接字,拆包等封装

test\ 测试用例/例子

close\		发送消息后保证消息送达后再断开连接

echo\		常见的pingpong测试， 最简单的例子

mgo\		mongodb异步读取例子

rpc\		异步远程过程调用

timer\		异步计时器

util\ 工具库

FAQ

问: 为什么接收消息回调必须需要手动转换类型, 例如: msg := content.(*gamedef.TestEchoACK), 而不是参数上写成参数类型?

答: cellnet这么设计是考虑到可以将参数进行多层传递, 如果写成不同消息类型, 传递就麻烦很多

这里鼓励消息注册者可以进行消息注册函数的封装, 在网关里, 就对socket.RegisterSessionMessage进行封装

func RegisterMessage(msgName string, userHandler func(interface{}, cellnet.Session, int64)) {

	msgMeta := cellnet.MessageMetaByName(msgName)

	if msgMeta == nil {
		log.Errorf("message register failed, %s", msgName)
		return
	}

	for _, conn := range routerConnArray {

		conn.RegisterCallback(msgMeta.ID, func(data interface{}) {

			if ev, ok := data.(*relayEvent); ok {

				rawMsg, err := cellnet.ParsePacket(ev.Packet, msgMeta.Type)

				if err != nil {
					log.Errorln("unmarshaling error:\n", err)
					return
				}

				msgContent := rawMsg.(interface {
					String() string
				}).String()				

				userHandler(rawMsg, ev.Ses, ev.ClientID)

			}

		})
	}

}

再来一个外层封装

func RegisterExternMessage(msgName string, userHandler func(interface{}, *Player)) {

	backend.RegisterMessage(msgName, func(content interface{}, routerSes cellnet.Session, clientid int64) {

		if player, ok := PlayerByID[clientid]; ok {

			userHandler(content, player)
		}
	})

}

Wiki

https://github.com/davyxu/cellnet/wiki

这里有文档和架构,设计解析

贡献者

欢迎提供dev分支的pull request

bug请直接通过issue提交

凡提交代码和建议, bug的童鞋, 均会在下列贡献者名单者出现

viwii(viwii@sina.cn), 提供一个可能造成死锁的bug

IronsDu(duzhongwei@qq.com), 大幅度性能优化

Chris Lonng(chris@lonng.org), 提供一个最大封包约束造成服务器间连接断开的bug

备注

感觉不错请star, 谢谢!

博客: http://www.cppblog.com/sunicdavy

知乎: http://www.zhihu.com/people/xu-bo-62-87

提交bug及特性: https://github.com/davyxu/cellnet/issues

贡献代码: https://github.com/davyxu/cellnet/pulls

Documentation

Overview

dispatcher包提供消息队列, 消息注册+派发 封装消息解包, 打包的过程

Index

• func BuildPacket(data interface{}) (*Packet, *MessageMeta)
• func ParsePacket(pkt *Packet, msgType reflect.Type) (interface{}, error)
• func RegisterMessageMeta(name string, msg proto.Message, id uint32)
• func VisitMessageMeta(callback func(*MessageMeta))
• type Connector
• type EventPipe
  • func NewEventPipe() EventPipe
• type EventQueue
• type MessageMeta
  • func MessageMetaByID(id uint32) *MessageMeta
  • func MessageMetaByName(name string) *MessageMeta
  • func MessageMetaByType(rtype reflect.Type) *MessageMeta
• type Packet
  • func (self Packet) ContextID() uint32
• type Peer
• type Session
• type SessionManager
• type Timer
  • func NewTimer(eq EventQueue, dur time.Duration, callback func(*Timer)) *Timer
  • func (self *Timer) Stop()

Functions

func BuildPacket

func BuildPacket(data interface{}) (*Packet, *MessageMeta)

消息到封包

func ParsePacket

func ParsePacket(pkt *Packet, msgType reflect.Type) (interface{}, error)

封包到消息

func RegisterMessageMeta

func RegisterMessageMeta(name string, msg proto.Message, id uint32)

注册消息元信息(代码生成专用)

func VisitMessageMeta

func VisitMessageMeta(callback func(*MessageMeta))

遍历消息元信息

Types

type Connector

type Connector interface {

	// 连接后的Session
	DefaultSession() Session

	// 自动重连间隔, 0表示不重连, 默认不重连
	SetAutoReconnectSec(sec int)
}

type EventPipe

type EventPipe interface {
	AddQueue() EventQueue

	Start()

	Stop(int)

	Wait() int

	// 开启捕获错误, 错误不会崩溃
	EnableCaputrePanic(enable bool)
}

func NewEventPipe

func NewEventPipe() EventPipe

type EventQueue

type EventQueue interface {

	// 注册事件回调
	RegisterCallback(id uint32, f func(interface{}))

	// 设置事件截获钩子, 在CallData中调用钩子
	InjectData(func(interface{}) bool)

	// 投递事件, 通过队列到达消费者端
	PostData(data interface{})

	// 直接调用消费者端的handler
	CallData(data interface{})

	// 延时投递
	DelayPostData(dur time.Duration, callback func())
}

type MessageMeta

type MessageMeta struct {
	Type reflect.Type
	Name string
	ID   uint32
}

func MessageMetaByID

func MessageMetaByID(id uint32) *MessageMeta

根据id查找消息元信息

func MessageMetaByName

func MessageMetaByName(name string) *MessageMeta

根据名字查找消息元信息

func MessageMetaByType

func MessageMetaByType(rtype reflect.Type) *MessageMeta

根据类型名字查找消息元信息

type Packet

type Packet struct {
	MsgID uint32 // 消息ID
	Data  []byte
}

普通封包

func (Packet) ContextID

func (self Packet) ContextID() uint32

type Peer

type Peer interface {

	// 开启
	Start(address string) Peer

	// 关闭
	Stop()

	// 名字
	SetName(string)
	Name() string

	// Session最大包大小, 超过这个数字, 接收视为错误, 断开连接
	SetMaxPacketSize(size int)
	MaxPacketSize() int

	// 事件
	EventQueue

	// 连接管理
	SessionManager
}

type Session

type Session interface {

	// 发包
	Send(interface{})

	// 直接发送封包
	RawSend(*Packet)

	// 断开
	Close()

	// 标示ID
	ID() int64

	// 归属端
	FromPeer() Peer
}

type SessionManager

type SessionManager interface {

	// 获取一个连接
	GetSession(int64) Session

	// 遍历连接
	VisitSession(func(Session) bool)

	// 连接数量
	SessionCount() int
}

type Timer

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

func NewTimer

func NewTimer(eq EventQueue, dur time.Duration, callback func(*Timer)) *Timer

func (*Timer) Stop

func (self *Timer) Stop()