daheige / cogo-job Goto Github PK

package gjobs

type BaseJob interface {
	Run(name string, args ...interface{})
}

package gjobs

import (
	"log"
	"time"
)

type HelloWorldJob struct {
}

func (job *HelloWorldJob) Run(jobName string, args ...interface{}) {
	if len(args) < 2 {
		log.Fatal("参数数量错误")
	}

	name := (args[0:1])[0] // 取第一个参数
	// 用timer来模拟一个需要运行30s的任务
	timer := time.After(time.Second * 6)
	select {
	case <-timer:
		break
	}
	log.Println("===============================")
	log.Println("===============================")
	log.Println("hello", name)
	log.Println("===============================")
	log.Println("===============================")
	log.Println("Job 运行完毕")
}

func execJob() {
	params := strings.Split(jobParamList, ";")
	jobObj := reflect.ValueOf(gjobs.AllJobMap[jobName])
	
	if jobObj == reflect.Zero(jobObj.Type()) {
		log.Println("找不到Job，退出执行")
		return
	}
	
	valueFunc := jobObj.MethodByName("Run")
	if valueFunc == reflect.Zero(valueFunc.Type()) {
		log.Println("找不到Job中的Run方法，退出执行")
		return
	}
	
	paramList := make([]reflect.Value, 0, len(params)+1) //多了一个jobName
	paramList = append(paramList, reflect.ValueOf(jobName))
	
	for k := range params {
		paramList = append(paramList, reflect.ValueOf(params[k]))
	}
	
	// 反射调用函数
	resultList := valueFunc.Call(paramList)
	log.Println("exec end", map[string]interface{}{
		"jobName":     jobName,
		"params":      params,
		"returnValue": resultList,
	})
	return
	
}

package gjobs

//please do not edit this JobMap.go, it is auto created by go generate
var AllJobMap = map[string]interface{}{
	"HelloWorldJob": &HelloWorldJob{},
}

...
//go:generate ./bin/init_job.sh
// 初始化一些工作，如启动参数获取，健康检查
func init() {
}
...

package main

import (
	"log"
	"time"
)

func Run(jobName string, args ...interface{}) {
	if len(args) < 2 {
		log.Fatal("参数数量错误", args)
	}
	
	name := (args[0:1])[0] // 取第一个参数
	// 用timer来模拟一个需要运行30s的任务
	timer := time.After(time.Second * 6)
	select {
	case <-timer:
		break
	}
	log.Println("===============================")
	log.Println("===============================")
	log.Println("This is a go plugin job demo", name)
	log.Println("===============================")
	log.Println("===============================")
	log.Println("Job 运行完毕")
}

func execJobWithPluginWay() {
	plugin, err := plugin.Open("./" + jobName + ".so")
	if err != nil {
		log.Fatal("找不到Job")
		return
	}
	// 查找是否有我们统一要求保留出来的Run方法
	runMethod, err := plugin.Lookup("Run")
	if err != nil {
		log.Fatal("找不到Job的Run方法", err)
		return
	}
	// 将我们通过main程序传进来的字符串参数转成interface{}变参
	params := strings.Split(jobParamList, ";")
	paramList := make([]interface{}, 0, len(params))
	for k := range params {
		paramList = append(paramList, params[k])
	}
	
	// 类型断言
	run := runMethod.(func(string, ...interface{}))
	// 执行Run方法
	run(jobName, paramList...)
	
	log.Println("exec end", map[string]interface{}{
		"jobName": jobName,
		"params":  params,
	})
	return
}

package main

import (
	"gowork/job"
	"os"
)

//go:generate ./bin/init_cobra_job.sh

func main() {
	if err := job.BakeJob.Execute(); err != nil {
		os.Exit(-1)
	}
}

var CogoJob = &cobra.Command{
	Use:               "cogo",
	Short:             "cogo, a job framework base on cobra",
	SilenceUsage:      true,
	DisableAutoGenTag: true,
	Long:              "基于cobra的Golang Job实现方案",
	PersistentPreRun:  preRun,
}

var coJobOptions = struct {
	pprofHost string // pprof 性能监控绑定的IP地址，正式环境尽量只允许本地访问
	pprofPort int    // pprof 性能监控随机获取一个可用的端口号，正式环境尽量只允许本地访问
	cpuNum    int    //指定cpu核数
	//TODO 在这里可以继续声明log目录，配置文件的路径等待
	//log_dir string
	//cfg_file string
}{}

func init() {
	//NOTE 这里不能使用h作为参数，-h参数为help占用
	CogoJob.PersistentFlags().StringVarP(&coJobOptions.pprofHost, "host", "m", "0.0.0.0", "pprof server ip")
	CogoJob.PersistentFlags().IntVarP(&coJobOptions.cpuNum, "cpu_num", "p", 1, "cpu num")
}

//preRun函数内的代码也可以放在init函数中进行
//      TODO 1. 初始化日志配置
//      TODO 2. 初始化数据库相关配置
//      TODO 3. 初始化Redis相关配置
//      TODO 4. etc...
func preRun(cmd *cobra.Command, args []string) {
	defer func() {
		<-pprofStarted
	}()

	// 开启pprof性能监控
	go func() {
		defer func() {
			if err := recover(); err != nil {
				log.Println("发生panic")
				lib.SendToUs("Job发生panic")
			}
		}()
		// 这些命令不需要启动pprof监控
		if cmd.Use == "version" || cmd.Use == "starter" {
			pprofStarted <- true
			return
		}
		address := fmt.Sprintf("%s:%d", coJobOptions.pprofHost, coJobOptions.pprofPort)
		mux := http.NewServeMux() //创建一个http ServeMux实例
		mux.HandleFunc("/debug/pprof/", pprof.Index)
		mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
		mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
		mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
		mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
		mux.HandleFunc("/check", HealthCheckHandler)
		mux.HandleFunc("/pid", GetThreadIdHandler)

		log.Println("server pprof run on: ", address)
		pprofStarted <- true

		if err := http.ListenAndServe(address, mux); err != nil {
			log.Fatal("pprof error: ", err)
		}
	}()

}

// 脚手架
var Starter = &cobra.Command{
	Use:   "starter",
	Short: "Start a new Job quickly",
	Long:  `Start a new Job quickly`,
	RunE:  startFunc,
}

// This file is auto generated.
package tasks

import (
	"fmt"
	"github.com/spf13/cobra"
)

var TestJob = &cobra.Command{
	Use:     "TestJob",
	Short:   "TestJob",
	Long:    "TestJob",
	Example: "./gobake TestJob", //TODO: write how to run your job
	RunE:    TestJobFunc,
}

// Job需要使用的参数，为了避免同包下参数重复定义，用Job名包裹起来
// TODO: If you need define local variables, implement your code here
//var testJobOptions struct {
//	age int
//	name string
//}

// This is **cobra.Command** runnable method, you can change method name,
// but avoid name conflict
func TestJobFunc(_ *cobra.Command, _ []string) error {
	//TODO: Implements your own job logic
	fmt.Println("Hello World")
	return nil
}