基于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)
    }
}

View Code

** master：**

主要负责接收后台对任务的管理提供api接口

go run .mastermainmaster.go -config= .mastermainmaster.json

** worker：**

监听任务，执行任务，任务调度

go run .workermainworker.go -config=./workermainworker.json -logDir=./logs/worker

该套分布式任务管理开箱即用，如有疑问可以随时提问

项目框架解析：供学习爱好者