golang+数据库定时任务

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了golang+数据库定时任务，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含4834字，纯文字阅读大概需要7分钟。

内容图文

golang+数据库定时任务

---

   项目背景大致如下，楼主在用nodejs写项目时遇到一些需要定时去处理的事情，例如僵尸用户定时清除，一些产品定时下架，邮件定时发送等等! 期初使用nodejs setTimeOut递归嵌套实现，后来发现内存不断飙升，故而放弃，最终改用了性能不错的golang实现

数据库设计

---

字段名称	含义
id	编号
name	任务名称
create_at	创建时间
type	1. 执行一次 2.循环执行
separate_time	执行间隔
status	执行状态 0.未开始 1. 执行中 -1.执行失败 -2.手动暂停
remark	备注信息
fn	要执行的数据库存储过程或函数
start_time	开始执行时间
next_exec_time	下次执行时间
last_exec_time	上次执行时间
fn_type	email, sql 等等

大致实现流程

---

1. 需要有一个死循环，sleep 10s启动然后sleep 10 …

    		for {
			time.Sleep(10 * time.Second)
			go execTask(*db) //使用子进程执行，防止卡死主进程
		}

1. 开始执行,查找需要执行的任务

rows, err := db.Query("SELECT id,name,status,type,fn,fn_type, separate_time FROM public.tasks where (status = 0 and start_time < now()) or (status = 1 and next_exec_time < now());")

1. 执行任务

res, err := db.Exec(fn)

1. 执行任务成功后，更新下次执行时间

func setTaskNextExecTime(db sql.DB, taskId string, separateTime int64) error {
	next_exec_time := time.Now().Unix() + separateTime
	nextTime := time.Unix(next_exec_time, 999)
	res, err := db.Exec("UPDATE tasks set status = 1, last_exec_time=now(), next_exec_time=$2 WHERE id = $1::uuid", taskId, nextTime)
	res = nil
	log.Println(res)
	return err;
}

优缺点

---

    优点：
        1. 所有任务执行状态都可以查询到，例如任务异常或者上次执行时间，下次执行时间
        2. 增加一个定时任务，只需要在数据库插入一条记录就OK
    缺点:
        1. 如果要绑定非数据库可操作任务，需要自己扩展
    

项目源码

---

// MTask project main.go
package main

import (
	"database/sql"
	_ "github.com/lib/pq"
	"log"
	"time"
	"os"
	"io/ioutil"
	"encoding/json"
)

//配置结构体
type Conf struct {
	Db map[string] string
}

//读取配置文件
func readConf(path string) (Conf, error) {
	var c Conf
	var err error
	
	fi, err := os.Open(path)
	if err != nil {
		return c, err 
	} else {
		defer fi.Close()
		
		//读取配置文件
		fd, err := ioutil.ReadAll(fi)
		if err != nil {
			return c, err
		} else {
			var c Conf
			err = json.Unmarshal(fd, &c)
			if err != nil {
				return c, err
			} else {
				return c, err
			}
		}
	}
	return c, err
}


func main() {
	c, err := readConf("./conf.json")
	if err != nil {
		log.Print(err)
		panic(err)
	}
	db, err := sql.Open("postgres", c.Db["postgres"])
	if err != nil {
		log.Print(err)
	} else {
		defer db.Close()
		for {
			time.Sleep(10 * time.Second)
			go execTask(*db)
		}
	}
}

func execTask(db sql.DB) {
	defer func() {
		if err := recover(); err != nil {
			log.Print(err)
			log.Printf("执行任务时发生错误:%s", err)
		}
	}();
	
	log.Println("开始执行任务.......")
	rows, err := db.Query("SELECT id,name,status,type,fn,fn_type, separate_time FROM public.tasks where (status = 0 and start_time < now()) or (status = 1 and next_exec_time < now());")
	if err != nil {
		log.Print(err)
	} else {
		defer rows.Close()
		for rows.Next() {
			var id string
			var name string
			var status int
			var taskType int
			var separateTime int64
			var fn string
			var fnType string

			err = rows.Scan(&id, &name, &status, &taskType, &fn, &fnType, &separateTime)

			if err != nil {
				//记录错误，同时更新任务信息为异常
				log.Print(err)
				err = setTaskExecFail(db, id)
				if err != nil {
					log.Print(err)
				}
			} else {
				if (fnType == "sql") {
					res, err := db.Exec(fn)
					if err != nil {
						log.Print(err)
						err = setTaskExecFail(db, id)
						if err != nil {
							log.Print(err)
						}
						log.Printf("任务:%s执行时出错", name)
					} else {
						res = nil
						log.Println(res)
						
						if taskType == 1 {
							err = setTaskExecSuccess(db, id)
							if err != nil {
								log.Print(err)
							}
							log.Printf("任务：%s执行完成", name)
						} else {
							err = setTaskNextExecTime(db, id, separateTime)
							if err != nil {
								log.Print(err)
							}
						}
						log.Printf("任务：%s执行成功", name)
					}
				} else if (fnType == "bash") {
					log.Printf("这是一个bash任务")
				} else if (fnType == "python") {
					log.Printf("这是一个python任务")
				} else if (fnType == "email") {
					//发送email任务
					err = ExecEmailTask(db)
					if err != nil {
						handleFail(db, id)
						log.Println(err)
					} else {
						handleSuccess(db, id)
					}
					log.Printf("发送邮件任务")
					setTaskExecSuccess(db, id)
					setTaskNextExecTime(db, id, separateTime)
				} else if (fnType == "sms") {
					//发送短信任务
					log.Printf("发送短信任务")
				}
				
			}
		}

		err = rows.Err()
		if err != nil {
			log.Print(err)
		}
	}
	log.Println("结束执行任务....")
}

func setTaskExecFail(db sql.DB, taskId string) error {
	res, err := db.Exec("UPDATE tasks set status = -2 WHERE id = $1::uuid", taskId)
	err = nil
	log.Println(res)
	return err
}

func setTaskExecSuccess(db sql.DB, taskId string) error {
	res, err := db.Exec("UPDATE tasks set status = 2 WHERE id = $1::uuid", taskId)
	err = nil
	log.Println(res)
	return err
}

func setTaskNextExecTime(db sql.DB, taskId string, separateTime int64) error {
	next_exec_time := time.Now().Unix() + separateTime
	nextTime := time.Unix(next_exec_time, 999)
	res, err := db.Exec("UPDATE tasks set status = 1, last_exec_time=now(), next_exec_time=$2 WHERE id = $1::uuid", taskId, nextTime)
	res = nil
	log.Println(res)
	return err;
}

golang+数据库定时任务

标签：

本文系统来源：http://my.oschina.net/wanglihui/blog/420261

内容总结

以上是互联网集市为您收集整理的golang+数据库定时任务全部内容，希望文章能够帮你解决golang+数据库定时任务所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---