go定时读取mysql_golang+数据库定时任务
golang+數(shù)據(jù)庫定時(shí)任務(wù)
項(xiàng)目背景大致如下，樓主在用nodejs寫項(xiàng)目時(shí)遇到一些需要定時(shí)去處理的事情，例如僵尸用戶定時(shí)清除，一些產(chǎn)品定時(shí)下架，郵件定時(shí)發(fā)送等等! 期初使用nodejs setTimeOut遞歸嵌套實(shí)現(xiàn)，后來發(fā)現(xiàn)內(nèi)存不斷飆升，故而放棄，最終改用了性能不錯(cuò)的golang實(shí)現(xiàn)
數(shù)據(jù)庫設(shè)計(jì)
字段名稱
含義
id
編號(hào)
name
任務(wù)名稱
create_at
創(chuàng)建時(shí)間
type
1. 執(zhí)行一次 2.循環(huán)執(zhí)行
separate_time
執(zhí)行間隔
status
執(zhí)行狀態(tài) 0.未開始 1. 執(zhí)行中 -1.執(zhí)行失敗 -2.手動(dòng)暫停
remark
備注信息
fn
要執(zhí)行的數(shù)據(jù)庫存儲(chǔ)過程或函數(shù)
start_time
開始執(zhí)行時(shí)間
next_exec_time
下次執(zhí)行時(shí)間
last_exec_time
上次執(zhí)行時(shí)間
fn_type
email, sql 等等
大致實(shí)現(xiàn)流程
需要有一個(gè)死循環(huán)，sleep 10s啟動(dòng)然后sleep 10 …
for {
time.Sleep(10 * time.Second)
go execTask(*db) //使用子進(jìn)程執(zhí)行，防止卡死主進(jìn)程
}
開始執(zhí)行,查找需要執(zhí)行的任務(wù)
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());")
執(zhí)行任務(wù)
res, err := db.Exec(fn)
執(zhí)行任務(wù)成功后，更新下次執(zhí)行時(shí)間
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;
}
優(yōu)缺點(diǎn)
優(yōu)點(diǎn)：
1. 所有任務(wù)執(zhí)行狀態(tài)都可以查詢到，例如任務(wù)異常或者上次執(zhí)行時(shí)間，下次執(zhí)行時(shí)間
2. 增加一個(gè)定時(shí)任務(wù)，只需要在數(shù)據(jù)庫插入一條記錄就OK
缺點(diǎn):
1. 如果要綁定非數(shù)據(jù)庫可操作任務(wù)，需要自己擴(kuò)展
項(xiàng)目源碼
// MTask project main.go
package main
import (
"database/sql"
_ "github.com/lib/pq"
"log"
"time"
"os"
"io/ioutil"
"encoding/json"
)
//配置結(jié)構(gòu)體
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("執(zhí)行任務(wù)時(shí)發(fā)生錯(cuò)誤:%s", err)
}
}();
log.Println("開始執(zhí)行任務(wù).......")
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 {
//記錄錯(cuò)誤，同時(shí)更新任務(wù)信息為異常
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("任務(wù):%s執(zhí)行時(shí)出錯(cuò)", name)
} else {
res = nil
log.Println(res)
if taskType == 1 {
err = setTaskExecSuccess(db, id)
if err != nil {
log.Print(err)
}
log.Printf("任務(wù)：%s執(zhí)行完成", name)
} else {
err = setTaskNextExecTime(db, id, separateTime)
if err != nil {
log.Print(err)
}
}
log.Printf("任務(wù)：%s執(zhí)行成功", name)
}
} else if (fnType == "bash") {
log.Printf("這是一個(gè)bash任務(wù)")
} else if (fnType == "python") {
log.Printf("這是一個(gè)python任務(wù)")
} else if (fnType == "email") {
//發(fā)送email任務(wù)
err = ExecEmailTask(db)
if err != nil {
handleFail(db, id)
log.Println(err)
} else {
handleSuccess(db, id)
}
log.Printf("發(fā)送郵件任務(wù)")
setTaskExecSuccess(db, id)
setTaskNextExecTime(db, id, separateTime)
} else if (fnType == "sms") {
//發(fā)送短信任務(wù)
log.Printf("發(fā)送短信任務(wù)")
}
}
}
err = rows.Err()
if err != nil {
log.Print(err)
}
}
log.Println("結(jié)束執(zhí)行任務(wù)....")
}
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;
}
有疑問加站長微信聯(lián)系(非本文作者)
《新程序員》：云原生和全面數(shù)字化實(shí)踐50位技術(shù)專家共同創(chuàng)作，文字、視頻、音頻交互閱讀總結(jié)
以上是生活随笔為你收集整理的go定时读取mysql_golang+数据库定时任务的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： mysql concat 去掉重复_my
- 下一篇： 调用方法[manageapp]时发生异常