tcp 服务端

type TcpServer struct {
// 记录每个IP的连接数量
perIPConnCounter ipmanager.PerIPConnCounter
ln               net.Listener
}

func(tc *TcpServer) Start() {
var lastPerIPErrorTime time.Time
//开始接收tcp连接请求
ln, err := net.Listen("tcp", ":9998")
if err != nil {
log.Println("Listen", err.Error())
return}
tc.ln = ln

// 接收并处理连接请求
for {
c, err := tc.acceptConn(ln, &lastPerIPErrorTime)
if err != nil {
panic("Room" + err.Error())
break
}
// 启动worker去处理连接
go tc.serve(c)
}
}

func(tc *TcpServer) acceptConn(ln net.Listener, lastPerIPErrorTime *time.Time) (net.Conn, error) {
for {
c, err := ln.Accept()
if err != nil {
if c != nil {
log.Println("[FATAL] net.Listener returned non-nil conn and non-nil error : ", err.Error())
}
if netErr, ok := err.(net.Error); ok && netErr.Temporary() {
log.Println("[ERROR] Temporary error when accepting new connections: ", err.Error())
time.Sleep(time.Second)
continue
}
if err != io.EOF && !strings.Contains(err.Error(), "use of closed network connection") {
log.Println("[ERROR] Permanent error when accepting new connections: ", err.Error())
return nil, err
}
return nil, io.EOF
}
if c == nil {
panic("BUG: net.Listener returned (nil, nil)")
}
//ConnectionTimeout 超时时间
c.SetReadDeadline(time.Now().Add(time.Duration(ConnectionTimeout) * time.Second))
//MaxConnSperIp 最大连接数
if MaxConnSperIp > 0 {
pic := tc.wrapPerIPConn(c)
if pic == nil {
if time.Since(*lastPerIPErrorTime) > time.Minute {
log.Println("[ERROR] The number of connections from ")
*lastPerIPErrorTime = time.Now()
}
continue
}
c = pic
}
return c, nil
}
}

func(tc *TcpServer) wrapPerIPConn(c net.Conn) net.Conn {
ip := ipmanager.GetUint32IP(c)
if ip == 0 {
return c
}
n := tc.perIPConnCounter.Register(ip)
//MaxConnSperIp 最大连接数
if n > MaxConnSperIp {
tc.perIPConnCounter.Unregister(ip)
c.Close()
return nil
}
return ipmanager.AcquirePerIPConn(c, ip, &tc.perIPConnCounter)
}

func(tc *TcpServer) serve(conn net.Conn) {
defer func() {
conn.Close()
if err := recover(); err != nil {
log.Println("readPacket", err)
return
}
}()
uc := NewUserClient(conn)
go readPacket(uc, uc.ChildStopChan)

for {
select {
case message, ok := <-uc.PacketDataChan:
if !ok {
log.Println("packet is get error")
continue
}
switch message.GetPacketType() {
case packet.APPLICATION_SEND:
//业务处理
log.Println("业务处理", message)
default:
log.Println("invalid messageReceived msg:", message)
break
}
case <-uc.LogOutStopChan:
goto LogOut
}
}
LogOut:
defer func() {
close(uc.ChildStopChan)
close(uc.LogOutStopChan)
close(uc.PacketDataChan)
}()
log.Println("LogOut close conn")
}

type UserClient struct {
Conn           net.Conn
ChildStopChan  chan bool
LogOutStopChan chan bool
PacketDataChan chan packet.Packet
}

func NewUserClient(conn net.Conn) *UserClient {
return &UserClient{Conn: conn,
ChildStopChan:   make(chan bool, 1),
LogOutStopChan:  make(chan bool, 1),
PacketDataChan:  make(chan packet.Packet, 100),
}
}

funcreadPacket(uc *UserClient, rStop <-chan bool) {
defer func() {
if err := recover(); err != nil {
log.Println("readPacket", err)
return
}
}()
for {
select {
case <-rStop:
log.Println("readPacket", "readPacket get stop cmd")
goto Stop
default:
pt, err := proto.ReadPacket(uc.Conn)
if err != nil {
log.Println("ReadPacket", err)
uc.LogOutStopChan <- true
goto Stop
}

//ConnectionTimeout 超时时间
uc.Conn.SetReadDeadline(time.Now().Add(time.Duration(ConnectionTimeout) * time.Second))
if pt != nil {
uc.PacketDataChan <- pt
}
}
}

Stop:
log.Println("conn stop ...")
}