go语言完善的基础设施为编写网络程序提供了极大的便利.只需要少量代码就可以编写一个高性能,稳定的异步网络程序.
本文介绍一个迷你的,基于事件回调的异步网络库.
首先简单介绍一下并发模型.
go提供了基于goroutine的同步网络接口,所以对每个网络连接可以创建一个单独的goroutine用于接收网络数据.这个goroutine是执行一个死循环,不断的recv数据,解包然后将完整的逻辑包发送到一个每连接唯一的chan中,供逻辑消费.
除了网络接收goroutine之外,每个连接还有一个专门的处理逻辑消息的goroutine,它的工作就是不断的从关联的chan中提取逻辑包并处理.
与之前的chuck-lua类似,为了让使用者可以方便定制自己的包结构,我提供了packet和decoder的抽象.
packet.go
package packet
const (
RAWPACKET = 1
RPACKET = 2
WPACKET = 3
EPACKET = 4
)
type Packet interface{
MakeWrite()(*Packet)
MakeRead() (*Packet)
Clone() (*Packet)
PkLen() (uint32)
DataLen() (uint32)
Buffer() (*ByteBuffer)
GetType() (byte)
}
decoder.go
package packet
import(
"net"
"encoding/binary"
"fmt"
"io"
)
var (
ErrPacketTooLarge = fmt.Errorf("Packet too Large")
ErrEOF = fmt.Errorf("Eof")
)
type Decoder interface{
DoRecv(Conn net.Conn)(Packet,error)
}
type RPacketDecoder struct{
maxpacket uint32
}
func NewRPacketDecoder(maxpacket uint32)(RPacketDecoder){
return RPacketDecoder{maxpacket:maxpacket}
}
func (this RPacketDecoder)DoRecv(Conn net.Conn)(Packet,error){
header := make([]byte,4)
n, err := io.ReadFull(Conn, header)
if n == 0 && err == io.EOF {
return nil,ErrEOF
}else if err != nil {
return nil,err
}
size := binary.LittleEndian.Uint32(header)
if size > this.maxpacket {
return nil,ErrPacketTooLarge
}
buf := make([]byte,size+4)
copy(buf[:],header[:])
n, err = io.ReadFull(Conn,buf[4:])
if n == 0 && err == io.EOF {
return nil,ErrEOF
}else if err != nil {
return nil,err
}
return NewRPacket(NewBufferByBytes(buf,(uint32)(len(buf)))),nil
}
type RawDecoder struct{
}
func NewRawDecoder()(RawDecoder){
return RawDecoder{}
}
func (this RawDecoder)DoRecv(Conn net.Conn)(Packet,error){
buff := make([]byte,4096)
n,err := Conn.Read(buff)
if n == 0 && err == io.EOF {
return nil,ErrEOF
}else if err != nil {
return nil,err
}
return NewRawPacket(NewBufferByBytes(buff,(uint32)(n))),nil
}
内置了2种包的类型,分别是rawpacket,rpacket/wpacket.
rawpacket其实就是原始二进制数据流,没有区分逻辑界限.
rpacket/wpacket则提供了一种4字节包头,的二进制流包结构.
eventpacket则作为内部使用,目前用来向逻辑通告连接关闭,错误等事件.
下面就是整个网络库的核心部分tcpsession,它提供了对tcp连接的处理.
package tcpsession
import(
"net"
packet "kendynet-go/packet"
"fmt"
)
var (
ErrUnPackError = fmt.Errorf("TcpSession: UnpackError")
ErrSendClose = fmt.Errorf("send close")
ErrSocketClose = fmt.Errorf("socket close")
)
type Tcpsession struct{
Conn net.Conn
Packet_que chan packet.Packet
decoder packet.Decoder
socket_close bool
ud interface{}
}
func (this *Tcpsession) SetUd(ud interface{}){
this.ud = ud
}
func (this *Tcpsession) Ud()(interface{}){
return this.ud
}
func dorecv(session *Tcpsession){
for{
p,err := session.decoder.DoRecv(session.Conn)
if session.socket_close{
break
}
if err != nil {
session.Packet_que <- packet.NewEventPacket(err)
break
}
session.Packet_que <- p
}
close(session.Packet_que)
}
func ProcessSession(tcpsession *Tcpsession,decoder packet.Decoder,
process_packet func (*Tcpsession,packet.Packet,error))(error){
if tcpsession.socket_close{
return ErrSocketClose
}
tcpsession.decoder = decoder
go dorecv(tcpsession)
for{
msg,ok := <- tcpsession.Packet_que
if !ok {
//log error
return nil
}
if packet.EPACKET == msg.GetType(){
process_packet(tcpsession,nil,msg.(packet.EventPacket).GetError())
}else{
process_packet(tcpsession,msg,nil)
}
if tcpsession.socket_close{
return nil
}
}
}
func NewTcpSession(conn net.Conn)(*Tcpsession){
session := new(Tcpsession)
session.Conn = conn
session.Packet_que = make(chan packet.Packet,1024)
session.socket_close = false
return session
}
func (this *Tcpsession)Send(wpk packet.Packet)(error){
if this.socket_close{
return ErrSocketClose
}
idx := (uint32)(0)
for{
buff := wpk.Buffer().Bytes()
end := wpk.PkLen()
n,err := this.Conn.Write(buff[idx:end])
if err != nil || n < 0 {
return ErrSendClose
}
idx += (uint32)(n)
if idx >= (uint32)(end){
break
}
}
return nil
}
func (this *Tcpsession)Close(){
if this.socket_close{
return
}
this.socket_close = true
this.Conn.Close()
}
代码十分简短只有一百行出头点,这里关键地方时,dorecv,Send和ProcessSession三个函数.
dorecv所做的就是不断调用decoder.DoRecv从网络中提取网络包,然后将其写入到Packet_que中.
Send函数则保证需要发送的数据被写入到内湖缓冲或出错才会返回.
ProcessSession则是整个库的核心所在,接收到新连接之后,用新建连接作为参数调用ProcessSession,当网络包到达或出错时将会回调使用者提供的回调函数.从实现看它只是简单的创建一个goroutine执行dorecv,然后在一个for循环中不断读取到达的网络包然后调用回调函数.
下面是一个使用示例,更多的示例请参考:https://github.com/sniperHW/kendynet-go
package main
import(
"net"
tcpsession "kendynet-go/tcpsession"
packet "kendynet-go/packet"
"fmt"
)
func main(){
service := ":8010"
tcpAddr,err := net.ResolveTCPAddr("tcp4", service)
if err != nil{
fmt.Printf("ResolveTCPAddr")
}
listener, err := net.ListenTCP("tcp", tcpAddr)
if err != nil{
fmt.Printf("ListenTCP")
}
for {
conn, err := listener.Accept()
if err != nil {
continue
}
session := tcpsession.NewTcpSession(conn)
fmt.Printf("a client comming\n")
go tcpsession.ProcessSession(session,packet.NewRawDecoder(),
func (session *tcpsession.Tcpsession,rpk packet.Packet,errno error){
if rpk == nil{
fmt.Printf("%s\n",errno)
session.Close()
return
}
session.Send(rpk)
})
}
}