2019獨角獸企業重金招聘Python工程師標準>>>

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 等等

大致實現流程

---

需要有一個死循環，sleep 10s啟動然后sleep 10 ...

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

開始執行,查找需要執行的任務

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());")

執行任務

res, err := db.Exec(fn)

執行任務成功后，更新下次執行時間

func setTaskNextExecTime(db sql.DB, taskId string, separateTime int64) error {next_exec_time := time.Now().Unix() + separateTimenextTime := 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 = nillog.Println(res)return err; }

優缺點

---

優點：1. 所有任務執行狀態都可以查詢到，例如任務異常或者上次執行時間，下次執行時間2. 增加一個定時任務，只需要在數據庫插入一條記錄就OK缺點:1. 如果要綁定非數據庫可操作任務，需要自己擴展

項目源碼

---

// MTask project main.go package mainimport ("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 Confvar err errorfi, 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 Conferr = 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 stringvar name stringvar status intvar taskType intvar separateTime int64var fn stringvar fnType stringerr = 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 = nillog.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 = nillog.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 = nillog.Println(res)return err }func setTaskNextExecTime(db sql.DB, taskId string, separateTime int64) error {next_exec_time := time.Now().Unix() + separateTimenextTime := 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 = nillog.Println(res)return err; }

轉載于:https://my.oschina.net/wanglihui/blog/420261

總結

以上是生活随笔為你收集整理的golang+数据库定时任务的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。