如何使用Go语言实现一个简单的异步任务框架呢?生产者消费者模型

suoniao 2021-01-28
需要:0索币

如何使用Go语言实现一个简单的异步任务框架呢?且听我一一道来。首先我们看一下常见的任务队列的架构:

任务队列架构图

所以作为一个任务队列,主要有这么几个部分组成:

  • Producer,负责把调用者的函数、参数等传入到broker里
  • Consumer,负责从broker里取出消息,并且消费,如果有持久化运行结果的需求,还需要进行持久化
  • 选择一个Producer和Consumer之间序列化和反序列化的协议

首先我们要定义一下Broker的接口:

type Broker interface {
        Acquire(string) *Task                                       
        Ack(*Task) bool                     
        Update(*Task)
        Enqueue(*Task) string
}

作为一个broker,获取任务,ACK,更新任务状态,入队是基本操作。然后我们要定义一个任务,既然是异步任务队列 嘛,这是自然的:

type Task struct {
        ID                  string    `json:"task_id"`
        CreatedAt           time.Time `json:"created_at"`
        UpdatedAt           time.Time `json:"updated_at"`
        QueueName           string    `json:"queue_name"`
        JobName             string    `json:"job_name"`
        ArgsMap             ArgsMap   `json:"args_map"`
        CurrentHandlerIndex int       `json:"current_handler_index"`
        OriginalArgsMap     ArgsMap   `json:"original_args_map"`
        ResultLog           string    `json:"result_log"`
}

接下来我们要选择一个序列化和反序列化的协议,为了通用,我们选择JSON。那么接下来要做的事情就比较简单了,我们实现broker的 四个接口即可,我们使用Redis的LPUSH和RBPOP就可以:

func (r *RedisBroker) Acquire(queueName string) *Task {
        task := Task{}
        vs, err := rc.BRPop(time.Duration(0), genQueueName(queueName)).Result()
        if err != nil {
                log.Panicf("failed to get task from redis: %s", err)
                return nil // never executed
        }
        v := []byte(vs[1])

        if err := json.Unmarshal(v, &task); err != nil {
                log.Panicf("failed to get task from redis: %s", err)
                return nil // never executed
        }

        return &task
}

func (r *RedisBroker) Ack(task *Task) bool {
        // redis doesn't support ACK
        return true
}

func (r *RedisBroker) Update(task *Task) {
        task.UpdatedAt = time.Now()
        taskBytes, err := json.Marshal(task)
        if err != nil {
                log.Panicf("failed to enquue task %+v: %s", task, err)
                return // never executed here
        }
        rc.Set(genTaskName(task.ID), taskBytes, time.Duration(r.TaskTTL)*time.Second)
}

func (r *RedisBroker) Enqueue(task *Task) string {
        taskBytes, err := json.Marshal(task)
        if err != nil {
                log.Panicf("failed to enquue task %+v: %s", task, err)
                return "" // never executed here
        }

        rc.Set(genTaskName(task.ID), taskBytes, time.Duration(r.TaskTTL)*time.Second)
        rc.LPush(genQueueName(task.QueueName), taskBytes)
        return task.ID
}

生产者该咋做呢?生产者要做的事情就是把Task序列化,然后推送到对应的列表里:

func (r *RedisBroker) Enqueue(task *Task) string {
	taskBytes, err := json.Marshal(task)
	if err != nil {
		log.Panicf("failed to enquue task %+v: %s", task, err)
		return "" // never executed here
	}

	rc.Set(genTaskName(task.ID), taskBytes, time.Duration(r.TaskTTL)*time.Second)
	rc.LPush(genQueueName(task.QueueName), taskBytes)
	return task.ID
}

到目前为止,我们已经确保了生产者能生产数据,然后放到redis里,消费者能从redis里消费数据。那么怎么实现异步任务呢? 还差一个步骤,那就是当从broker里取到数据的时候,我们要能够找到对应的函数去执行:

func Register(jobName string, handlers ...JobHandler) {
	if _, ok := jobMap[jobName]; ok {
		log.Panicf("job name %s already exist, check your code", jobName)
		return // never executed here
	}

	jobMap[jobName] = handlers
}

所以我们需要这个函数,我们把执行的函数和一个唯一键对应起来,然后再在消费端,取到任务之后,把函数找到然后去执行:

handlers, ok := jobMap[task.JobName]
if !ok {
    log.Panicf("can't find job handlers of %s", task.JobName)
    return
}

大功告成!详细的代码看 这里

这样我们就实现了一个异步任务框架,如果去看看的话,你会发现 4k star 的 machinery 也是差不多的思路和代码。


参考资料:

回帖
  • 消灭零回复
相关主题
golang实现内存池 go语言字节池byte pool实现代码 0
golang网络编程之基于TCP协议实现长连接 golang心跳检测 0
golang类型断言type的使用 0
golang依赖包管理 mod使用教程 0
golang表单验证库validator 0
如何使用Go语言实现一个简单的异步任务框架呢?生产者消费者模型 0
golang 网络编程设置keepAlive空闲多长时间开始探测、 探测总次数 0
golang网络编程之TCP编程详解 0
go语言利用ioutil.ReadAll读取TCP socket所有数据 0
golang利用io.copy和bytes.Buffer读取TCP socket所有数据 0
golang读取所有socket数据的方式一 0
围绕Handler接口的方法ServeHTTP,可以轻松的写出go中的中间件 0
golang语言错误处理方式check函数,把错误转化为panic 0
golang cannot find module providing package 0
golang语言错误处理errors包使用详解 0
golang基于通道channel实现一个通用连接池 pool 0
golang数据类型之map结构详解 0
golang socket关闭读导致 wsarecv: An existing connection was forcibly closed by the remote host 0
windows配置golang的环境变量 gopath配置学习go语言第一天 0
golang应该是可以代替对使用c++存在误解的人吧,代替语言是不可能的 0
相关主题
QListWidget滚动的时候显示不完整 滚动条模式导致的哦 0
Qt设置顶层面板背景透明Qt::WA_TranslucentBackground 隐藏边框Qt::FramelessWindowHint 0
QMenu和QMenuBar样式表大全 qss 0
QT定时器startTimer和timerEvent事件 每隔interval 毫秒就会启动一次 0
C语言内存分配函数malloc和calloc的区别 0
QCheckBox的QSS样式表总结 0
Qt通过qRegisterMetaType注册自定义数据类型 0
QListwidget触发2次itemClicked事件 0
打印机USB驱动开发之实现打印服务器 0
Qt利用QLabel组件来显示图片 0
TableView自定义代理QStyledItemDelegate实现ComboBox 0
Qt利用QGraphicsView类实现图片放大缩小平移显示 0
Qt实现非阻塞延迟方法sleep 0
海康相机SDK的C++对应的接口 0
Qt实现webdav客户端功能支持https协议的webdav客户端 0
CHKDSK解决 移动硬盘只能看见盘符其它信息都看不见另外双击也打不开 0
gogs一直报errror:dial tcp xxx.xxx.xxx.xxx 宿主机的ip 0
索鸟快传2.1.2发布 0
索鸟快传2.1.1发布 0
Qt操作windows注册表的方法 bat从注册表中将键值删除 0