基於golang的分布式任務管理系統

先看看實現后的具體后台頁面

 

 

 

 

 

 

 

 

 

后台對定時任務的curd暫時不開放源碼，后台php開發的，沒啥難度，主要是像etcd中對任務進行操作（后台公司再用，抱歉不能公開后台源碼），后台請求的api接口在master目錄中；

 

學習golang的分布式任務之前先學習協etcd的使用

etcd的安裝

etcd的使用

 

重點需要學習etcd的  續租，watch監聽key

 

 

 

 golang分布式任務主要分為

master 服務端   主要提供后台操作的api，添加定時任務和一次性任務到etcd中，對任務進行增刪改查 日志  服務發現 服務注冊等功能

worker 客戶端   主要監聽etcd中的任務（與自己服務器ip相關的任務） 一次性任務立即執行即可，定時任務解析表達式后，放入內存中，定時執行

 

woker執行流程：

 

 githu地址：https://github.com/sunlongv520/go-crontab

 

 

一下code可以手動向etcd添加任務 或者刪除任務

package main

import (
    "context"
    "fmt"
    "github.com/coreos/etcd/clientv3"
    "time"
)

func main(){
    var (
        config clientv3.Config
        err error
        client *clientv3.Client
        kv clientv3.KV
        getResp *clientv3.GetResponse

    )
    //配置
    config = clientv3.Config{
        Endpoints:[]string{"192.168.2.232:2379"},
        DialTimeout:time.Second*5,
    }
    //連接 床見一個客戶端
    if client,err = clientv3.New(config);err != nil{
        fmt.Println(err)
        return
    }


    //用於讀寫etcd的鍵值對
    kv = clientv3.NewKV(client)

    //刪除key

    //kv.Delete(context.TODO(),"/cron/jobs/192.168.2.246/job1",clientv3.WithPrefix())
    //kv.Delete(context.TODO(),"/cron/jobs",clientv3.WithPrefix())
    //kv.Delete(context.TODO(),"/cron/oncejobs",clientv3.WithPrefix())
    //
    //return


    //新增定時任務
    //putResp, err := kv.Put(context.TODO(),"/cron/jobs/192.168.2.232/job2","{\"name\":\"job2\",\"command\":\"D:/phpstudy/PHPTutorial/php/php-5.6.27-nts/php E:/WWW/a.php\",\"cronExpr\":\"*/7 * * * * * *\"}",clientv3.WithPrevKV())
    //putResp, err := kv.Put(context.TODO(),"/cron/jobs/192.168.2.246/job2","{\"name\":\"job2\",\"command\":\" echo hello world\",\"cronExpr\":\"*/5 * * * * * *\"}",clientv3.WithPrevKV())
    //putResp, err := kv.Put(context.TODO(),"/cron/jobs/192.168.2.246/job3","{\"name\":\"job3\",\"command\":\" echo hello boy\",\"cronExpr\":\"*/10 * * * * * *\"}",clientv3.WithPrevKV())
//fmt.Println(putResp)
//fmt.Println(err)

    //新增一次性任務
    //putResp, err := kv.Put(context.TODO(),"/cron/oncejobs/192.168.2.232/job10","{\"name\":\"job10\",\"command\":\" echo hello world  \"}",clientv3.WithPrevKV())


    //強殺任務
    //putResp, err := kv.Put(context.TODO(),"/cron/killer/192.168.2.246/job10","")

    //
    //if err != nil{
    //    fmt.Println(err)
    //}else{
    //    fmt.Println("Revision:",putResp.Header.Revision)
    //    if putResp.PrevKv != nil{
    //        fmt.Println("key:",string(putResp.PrevKv.Key))
    //        fmt.Println("Value:",string(putResp.PrevKv.Value))
    //        fmt.Println("Version:",string(putResp.PrevKv.Version))
    //    }
    //}

    //查詢
    getResp,err = kv.Get(context.TODO(),"/cron/workers",clientv3.WithPrefix())
    if err != nil {
        fmt.Println(err)
        return
    }
    for _, kvpair := range getResp.Kvs {
        fmt.Println(kvpair)
    }


    getResp,err = kv.Get(context.TODO(),"/cron/jobs",clientv3.WithPrefix())
    if err != nil {
        fmt.Println(err)
        return
    }
    for _, kvpair := range getResp.Kvs {
        fmt.Println(kvpair)
    }

    getResp,err = kv.Get(context.TODO(),"/cron/oncejobs",clientv3.WithPrefix())
    if err != nil {
        fmt.Println(err)
        return
    }
    for _, kvpair := range getResp.Kvs {
        fmt.Println(kvpair)
    }
}

 

** master：**

主要負責接收后台對任務的管理 提供api接口

go run .\master\main\master.go -config= .\master\main\master.json

** worker：**

監聽任務，執行任務，任務調度

go run .\worker\main\worker.go -config=./worker\main\worker.json -logDir=./logs/worker

 

該套分布式任務管理開箱即用，如有疑問可以隨時提問

 

項目框架解析： 供學習愛好者

 

#!/bin/bash
cd /data/gocode/golang_asynctask
git reset --hard HEAD
git pull origin master
filename="/data2/gocode/golang_asynctask/build/crm/sync_comuser_to_erp"
rm -f $filename
go build -o /data2/gocode/golang_asynctask/build/crm/sync_comuser_to_erp /data2/gocode/golang_asynctask/cmd/crm/sync_comuser_to_erp/main.go
supervisorctl restart crm_sync_comuser_to_erp:*