async_task_demon

README

异步任务管理器

事件回调模式的异步任务

工作原理

• 任务管理者(有默认实现)
• 任务执行者
• 任务(有默认实现)

注意

• 大并发操作DB是,一定注意链接数问题
• 常用的orm,经测试,只有xorm能锁定链接数

测试异步写库

• orm: xorm,1000的链接数
• 总任务:100000
• 并行数:1000
• 任务执行者:40次写入
• 用进:261秒,写入4000000条记录

// AsyncTaskManagerInterface  异步任务管理器通用接口
type AsyncTaskManagerInterface interface {
	Start()                                                                              // 开始
	Stop()                                                                               // 停止
	SetWaitListMaxLength(l int)                                                          // 设置等待队列的最大长度,默认:1000
	SetWorkPoolSize(s int)                                                               // 设置并行的数量,默认:100
	AddAsyncTask(t AsyncTaskInterface, h AsyncTaskHandle, cb ...AsyncTaskCallBack) error // 增加一个异步任务
	SetFinishCallBack(f func(*TaskManagerResult))                                        // 设置任务完成的回调
	CheckTaskInWork(tag string) (bool, error)                                            // 查验任务是否在执行
	GetWaitLength() int                                                                  // 获取当前等待队列长度
	GetWorkerLength() int                                                                // 获取当前工作队列长度
	GetManagerStatus() uint8                                                             // 获取管理器状态
	SetTaskMaxLength(s int)                                                              // 设置最大任务数
	SetProgressCallBack(p int, cb ProgressCallBack)                                      // 设置进度回调
}

type AsyncTaskInterface interface {
	GetTag() *TaskTag       // 任务全局唯一标识
	SetStartUnix(end int64) // 设置任务开始时间
	GetStartUnix() int64    // 获取任务开始时间
	SetEndUnix(end int64)   // 设置任务结束时间
	GetEndUnix() int64      // 获取任务结束时间
	SetStatus(s uint8)      // 设置当前任务状态
	GetStatus() uint8       // 获取当前任务状态
	SetError(e error)
	GetError() error
}

type AsyncTaskHandle interface {
	Start(AsyncTaskInterface)              // 开始执行任务
	Stop()                                 // 结束任务,然后拉起回调,返回enmu.UserStopError
	SetHandleCallBack(f AsyncTaskCallBack) // 配置执行的回调
}

// AsyncTaskCallBack  异步任务回调
type AsyncTaskCallBack func(AsyncTaskInterface)

// TaskManagerResult   任务管理器执行结果
type TaskManagerResult struct {
	StartTime int64
	EndTime   int64
	Length    uint64
	TagsMap   map[string]AsyncTaskInterface
	Status    uint8
}

// ProgressCallBack   进度回调
type ProgressCallBack func(*TaskProgress)

// TaskProgress    进度信息
type TaskProgress struct {
	WorkNum    int   // 任务并行数
	AllTaskNum int   // 总任务数,如果没有SetTaskMaxLength(),返回0
	Progress   int   // 当前完成个数
	ParentTime int64 // 上次进度回调的时间
}