前言

你好，我是醉墨居士，我们上篇博客实现了路由分组的功能，接下来这篇博客我们将要实现任务管理模块

接口

trait/task_mgr.go

type TaskMgr interface {
	RouterGroup

	Start()
	StartWorker(taskQueue chan Request)
	Submit(request Request)
}

结构体

gcore/task_mgr.go

// TaskMgr 任务管理器
type TaskMgr struct {
	trait.RouterGroup

	taskQueues []chan trait.Request
}

// NewTaskMgr 创建任务管理器
func NewTaskMgr() trait.TaskMgr {
	taskQueues := make([]chan trait.Request, global.Config.TaskQueues)
	for i := 0; i < len(taskQueues); i++ {
		taskQueues[i] = make(chan trait.Request, global.Config.TaskQueueLen)
	}

	// 新建任务处理路由器与分组路由
	router := NewRouter()
	routerGroup := NewRouterGroup(router)

	return &TaskMgr{
		RouterGroup: routerGroup,
		taskQueues: taskQueues,
	}
}

接口实现

gcore/task_mgr.go

// Start 启动任务管理器
func (m *TaskMgr) Start() {
	for i := 0; i < len(m.taskQueues); i++ {
		for j := 0; j < global.Config.WorkersPerTaskQueue; j++ {
			go m.StartWorker(m.taskQueues[i])
		}
	}
}

// StartWorker 启动任务消费者
func (m *TaskMgr) StartWorker(taskQueue chan trait.Request) {
	for request := range taskQueue {
		flow := m.TaskFlow(request.ID())
		ctx := NewContext(request, flow)
		ctx.Next()
	}
}

// Submit 提交任务
func (m *TaskMgr) Submit(request trait.Request) {
	m.taskQueues[int(request.ConnID()) % len(m.taskQueues)] <- request
}

// Use 注册插件
func (m *TaskMgr) Use(flow ...trait.TaskFunc) {
	m.RouterGroup.Use(flow...)
}

// Regist 注册任务流
func (m *TaskMgr) Regist(id uint16, flow ...trait.TaskFunc) {
	m.RouterGroup.Regist(id, flow...)
}

// Regist 注册任务流
func (m *TaskMgr) RegistFlow(id uint16, flow trait.TaskFlow) {
	m.RouterGroup.RegistFlow(id, flow)
}

最后

我是醉墨居士，我们完成了基本的任务管理器的开发，希望对你有所帮助，也希望你有所收获