前言

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

接口

trait/task_mgr.go

type TaskMgr interface {RouterGroupStart()StartWorker(taskQueue chan Request)Submit(request Request)
}

结构体

gcore/task_mgr.go

// TaskMgr 任务管理器
type TaskMgr struct {trait.RouterGrouptaskQueues []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)
}

项目地址

Github：https://github.com/zm50/gte
Giee：https://gitee.com/zm50/gte

最后

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