Ainx的多路由模式

📕作者简介： 过去日记，致力于Java、GoLang,Rust等多种编程语言，热爱技术，喜欢游戏的博主。
📗本文收录于Ainx系列，大家有兴趣的可以看一看
📘相关专栏Rust初阶教程、go语言基础系列、spring教程等，大家有兴趣的可以看一看
📙Java并发编程系列，设计模式系列、go web开发框架系列正在发展中，喜欢Java，GoLang，Rust，的朋友们可以关注一下哦！

📙 本文大部分都是借鉴刘丹冰大佬的zinx框架和文章，更推荐大家去读大佬的原文,本文只是个人学习的记录

文章目录

Ainx的多路由模式
- 创建消息管理模块
- - 创建消息管理模块抽象类
  - 实现消息管理模块
- Ainx-V0.6代码实现
- 使用Ainx-V0.6完成应用程序

Ainx的多路由模式

我们之前在已经给Zinx配置了路由模式，但是很惨，之前的Zinx好像只能绑定一个路由的处理业务方法。显然这是无法满足基本的服务器需求的，那么现在我们要在之前的基础上，给Zinx添加多路由的方式。

Apis map[uint32] ziface.IRouter

这里起名字是Apis，其中key就是msgId， value就是对应的Router，里面应是使用者重写的Handle等方法。

那么这个Apis应该放在哪呢。

我们再定义一个消息管理模块来进行维护这个Apis。

创建消息管理模块

创建消息管理模块抽象类

在ainx/ainterface下创建imsghandler.go文件

/*
消息管理抽象层
*/
type IMsgHandle interface {
	DoMsgHandler(request IRequest)          //马上以非阻塞方式处理消息
	AddRouter(msgId uint32, router IRouter) //为消息添加具体的处理逻辑
}

这里面有两个方法，AddRouter()就是添加一个msgId和一个路由关系到Apis中，那么DoMsgHandler()则是调用Router中具体Handle()等方法的接口。

实现消息管理模块

在ainx/anet下创建msghandler.go文件。

package anet

import (
	"ainx/ainterface"
	"fmt"
	"strconv"
)

type MsgHandle struct {
	Apis map[uint32]ainterface.IRouter //存放每个MsgId 所对应的处理方法的map属性
}

func NewMsgHandle() *MsgHandle {
	return &MsgHandle{
		Apis: make(map[uint32]ainterface.IRouter),
	}
}

// 马上以非阻塞方式处理消息
func (mh *MsgHandle) DoMsgHandler(request ainterface.IRequest) {
	handler, ok := mh.Apis[request.GetMsgID()]
	if !ok {
		fmt.Println("api msgId = ", request.GetMsgID(), " is not FOUND!")
		return
	}

	//执行对应处理方法
	handler.PreHandle(request)
	handler.Handle(request)
	handler.PostHandle(request)
}

// 为消息添加具体的处理逻辑
func (mh *MsgHandle) AddRouter(msgId uint32, router ainterface.IRouter) {
	//1 判断当前msg绑定的API处理方法是否已经存在
	if _, ok := mh.Apis[msgId]; ok {
		panic("repeated api , msgId = " + strconv.Itoa(int(msgId)))
	}
	//2 添加msg与api的绑定关系
	mh.Apis[msgId] = router
	fmt.Println("Add api msgId = ", msgId)
}

Ainx-V0.6代码实现

首先iserver的AddRouter()的接口要稍微改一下，增添MsgId参数

ainx/ainterface/iserver.go

package ainterface

// 定义服务器接口
type IServer interface {
	//启动服务器方法
	Start()
	//停止服务器方法
	Stop()
	//开启业务服务方法
	Serve()
	//路由功能：给当前服务注册一个路由业务方法，供客户端链接处理使用
	AddRouter(msgId uint32, router IRouter)
	// todo 路由分组 未来目标 添加类似hertz Group分组，为每个链接分组
}

其次，Server类中之前有一个Router成员，代表唯一的处理方法，现在应该替换成MsgHandler成员

ainx/anet/server.go


type Server struct {
	// 设置服务器名称
	Name string
	// 设置网络协议版本
	IPVersion string
	// 设置服务器绑定IP
	IP string
	// 设置端口号
	Port string
	//当前Server的消息管理模块，用来绑定MsgId和对应的处理方法
	msgHandler ainterface.IMsgHandle
	//todo 未来目标提供更多option字段来控制server实例化
}

初始化Server自然也要更正，增加msgHandler初始化

/*
创建一个服务器句柄
*/
func NewServer() ainterface.IServer {
	//先初始化全局配置文件
	utils.GlobalSetting.Reload()

	s := &Server{
		Name:       utils.GlobalSetting.Name, //从全局参数获取
		IPVersion:  "tcp4",
		IP:         utils.GlobalSetting.Host,    //从全局参数获取
		Port:       utils.GlobalSetting.TcpPort, //从全局参数获取
		msgHandler: NewMsgHandle(),
	}
	return s
}

然后当Server在处理conn请求业务的时候，创建conn的时候也需要把msgHandler作为参数传递给Connection对象

//...
dealConn := NewConntion(conn, cid, s.msgHandler)
//...

那么接下来就是Connection对象了。固然在Connection对象中应该有MsgHandler的成员，来查找消息对应的回调路由方法

ainx/anet/connection.go


type Connection struct {
	//当前链接的socket TCP套接字
	Conn *net.TCPConn
	// 当前链接的ID也可以称作SessionID，ID全局唯一
	ConnID uint32
	// 当前链接的关闭状态
	isClosed bool

	//消息管理MsgId和对应处理方法的消息管理模块
	MsgHandler ainterface.IMsgHandle

	// 告知该链接已经退出/停止的channel
	ExitBuffChan chan bool
}
// 创建链接的方法
func NewConnection(conn *net.TCPConn, connID uint32, msgHandler ainterface.IMsgHandle) *Connection {
	c := &Connection{
		Conn:         conn,
		ConnID:       connID,
		isClosed:     false,
		MsgHandler:   msgHandler,
		ExitBuffChan: make(chan bool, 1),
	}
	return c
}

最后，在conn已经拆包之后，需要调用路由业务的时候，我们只需要让conn调用MsgHandler中的DoMsgHander()方法就好了

ainx/anet/connection.go


// 处理conn读数据的Goroutine
func (c *Connection) StartReader() {
	fmt.Println("Reader Goroutine is  running")
	defer fmt.Println(c.RemoteAddr().String(), " conn reader exit!")
	defer c.Stop()

	for {
		// 创建拆包解包的对象
		dp := NewDataPack()

		//读取客户端的Msg head
		headData := make([]byte, dp.GetHeadLen())
		if _, err := io.ReadFull(c.GetTCPConnection(), headData); err != nil {
			fmt.Println("read msg head error ", err)
			c.ExitBuffChan <- true
			continue
		}

		//拆包，得到msgid 和 datalen 放在msg中
		msg, err := dp.Unpack(headData)
		if err != nil {
			fmt.Println("unpack error ", err)
			c.ExitBuffChan <- true
			continue
		}

		//根据 dataLen 读取 data，放在msg.Data中
		var data []byte
		if msg.GetDataLen() > 0 {
			data = make([]byte, msg.GetDataLen())
			if _, err := io.ReadFull(c.GetTCPConnection(), data); err != nil {
				fmt.Println("read msg data error ", err)
				c.ExitBuffChan <- true
				continue
			}
		}
		msg.SetData(data)

		//得到当前客户端请求的Request数据
		req := Request{
			conn: c,
			msg:  msg, //将之前的buf 改成 msg
		}
		//从路由Routers 中找到注册绑定Conn的对应Handle
		go c.MsgHandler.DoMsgHandler(&req)
	}
}

使用Ainx-V0.6完成应用程序

Server.go

package main

import (
	"ainx/ainterface"
	"ainx/anet"
	"fmt"
)

// ping test 自定义路由
type PingRouter struct {
	anet.BaseRouter
}

// Ping Handle
func (this *PingRouter) Handle(request ainterface.IRequest) {
	fmt.Println("Call PingRouter Handle")
	//先读取客户端的数据，再回写ping...ping...ping
	fmt.Println("recv from client : msgId=", request.GetMsgID(), ", data=", string(request.GetData()))

	err := request.GetConnection().SendMsg(0, []byte("ping...ping...ping"))
	if err != nil {
		fmt.Println(err)
	}
}

// HelloZinxRouter Handle
type HelloZinxRouter struct {
	anet.BaseRouter
}

func (this *HelloZinxRouter) Handle(request ainterface.IRequest) {
	fmt.Println("Call HelloZinxRouter Handle")
	//先读取客户端的数据，再回写ping...ping...ping
	fmt.Println("recv from client : msgId=", request.GetMsgID(), ", data=", string(request.GetData()))

	err := request.GetConnection().SendMsg(1, []byte("Hello Ainx Router V0.6"))
	if err != nil {
		fmt.Println(err)
	}
}

func main() {
	//创建一个server句柄
	s := anet.NewServer()

	//配置路由
	s.AddRouter(0, &PingRouter{})
	s.AddRouter(1, &HelloZinxRouter{})

	//开启服务
	s.Serve()
}

Server端设置了2个路由，一个是MsgId为0的消息会执行PingRouter{}重写的Handle()方法，一个是MsgId为1的消息会执行HelloZinxRouter{}重写的Handle()方法。

我们现在写两个客户端，分别发送0消息和1消息来进行测试Zinx是否能够处理2个不同的消息业务。

Client.go

package main

import (
	"ainx/anet"
	"fmt"
	"io"
	"net"
	"time"
)

/*
模拟客户端
*/
func main() {

	fmt.Println("Client Test ... start")
	//3秒之后发起测试请求，给服务端开启服务的机会
	time.Sleep(3 * time.Second)

	conn, err := net.Dial("tcp", "127.0.0.1:8080")
	if err != nil {
		fmt.Println("client start err, exit!")
		return
	}

	for {
		//发封包message消息
		dp := anet.NewDataPack()
		msg, _ := dp.Pack(anet.NewMsgPackage(0, []byte("Ainx V0.6 Client0 Test Message")))
		_, err := conn.Write(msg)
		if err != nil {
			fmt.Println("write error err ", err)
			return
		}

		//先读出流中的head部分
		headData := make([]byte, dp.GetHeadLen())
		_, err = io.ReadFull(conn, headData) //ReadFull 会把msg填充满为止
		if err != nil {
			fmt.Println("read head error")
			break
		}
		//将headData字节流 拆包到msg中
		msgHead, err := dp.Unpack(headData)
		if err != nil {
			fmt.Println("server unpack err:", err)
			return
		}

		if msgHead.GetDataLen() > 0 {
			//msg 是有data数据的，需要再次读取data数据
			msg := msgHead.(*anet.Message)
			msg.Data = make([]byte, msg.GetDataLen())

			//根据dataLen从io中读取字节流
			_, err := io.ReadFull(conn, msg.Data)
			if err != nil {
				fmt.Println("server unpack data err:", err)
				return
			}

			fmt.Println("==> Recv Msg: ID=", msg.Id, ", len=", msg.DataLen, ", data=", string(msg.Data))
		}

		time.Sleep(1 * time.Second)
	}
}