cleanjob

README

cleanjob

一个golang实现的分布式计划任务内库，依托于 "github.com/robfig/cron/v3"的单机实现扩展了一个分布式实现

目录

• 安装
• Example

Install

Install the package with:

go get github.com/larry-dev/cleanjob

Import it with:

import "github.com/larry-dev/cleanjob"

Example

Please check the example folder for details.

package main

import (
	"flag"
	"fmt"
	"github.com/gin-gonic/gin"
	"github.com/larry-dev/cleanjob"
	"time"
)

var (
	addr string
)

func init() {
	flag.StringVar(&addr, "addr", ":8080", "服务端口号")
	flag.Parse()
}
type TestJob struct {
}
func (j *TestJob) Execute(ex *cleanjob.ExecuteContext) error {
	fmt.Println(ex.Job.String())
	time.Sleep(10 * time.Second)
	fmt.Println("执行完成")
	return nil
}

type Req struct {
	Ex   string `json:"ex"`
	Key  string `json:"key"`
	Data string `json:"data"`
}
func main() {
	men, err := cleanjob.NewRedisStore("127.0.0.1:6379")
	if err != nil {
		panic(err)
	}
	mange := cleanjob.NewManage(men)

	mange.On("game", &TestJob{})
	app := gin.Default()
	app.POST("job/add", func(c *gin.Context) {
		var req Req
		if err := c.BindJSON(&req); err != nil {
			return
		}
		job := cleanjob.NewJob(req.Ex, req.Key)
		job.WithData([]byte(req.Data))
		job.WithCronTime(time.Now().Add(10 * time.Second))
		if err := mange.Save(job); err != nil {
			c.JSON(500, gin.H{
				"err": err.Error(),
			})
			return
		}
		c.JSON(200, gin.H{})
	})
	app.POST("job/del", func(c *gin.Context) {
		var req Req
		if err := c.BindJSON(&req); err != nil {
			return
		}
		job := cleanjob.NewJob(req.Ex, req.Key)
		job.WithData([]byte(req.Data))
		job.WithCronTime(time.Now().Add(10 * time.Second))
		if err := mange.Delete(job); err != nil {
			c.JSON(500, gin.H{
				"err": err.Error(),
			})
			return
		}
		c.JSON(200, gin.H{})
	})
	mange.Run()
	app.Run(addr)
}

Documentation

Index

• Variables
• type Execute
• type ExecuteContext
• type IStore
• type Job
  • func NewJob(execute, key string) *Job
  • func (j *Job) String() string
  • func (j *Job) Struct(bean interface{})
  • func (j *Job) WithCronTime(t time.Time)
  • func (j *Job) WithData(data []byte)
  • func (j *Job) WithStructToData(data interface{})
• type Manage
  • func NewManage(store IStore) *Manage
  • func (m *Manage) Delete(job *Job) error
  • func (m *Manage) Load() ([]*Job, error)
  • func (m *Manage) On(key string, execute Execute)
  • func (m *Manage) Run()
  • func (m *Manage) Save(job *Job) error
• type Memory
  • func NewMemory() *Memory
  • func (m *Memory) Delete(job *Job) error
  • func (m *Memory) Load() ([]*Job, error)
  • func (m *Memory) Lock(ex *ExecuteContext) error
  • func (m *Memory) Save(job *Job) error
  • func (m *Memory) Test(job *Job)
  • func (m *Memory) UnLock(ex *ExecuteContext) error
  • func (m *Memory) Watch(f WatchFunc)
• type RedisStore
  • func NewRedisStore(address string, options ...redis.DialOption) (*RedisStore, error)
  • func (s *RedisStore) Delete(job *Job) (err error)
  • func (s *RedisStore) Load() ([]*Job, error)
  • func (s *RedisStore) Lock(ex *ExecuteContext) error
  • func (s *RedisStore) Save(job *Job) error
  • func (s *RedisStore) UnLock(ex *ExecuteContext) error
  • func (s *RedisStore) Watch(f WatchFunc)
• type Scheduler
  • func (s *Scheduler) Next(t time.Time) time.Time
  • func (s *Scheduler) Run()
• type WatchFunc

Variables

var (
	AddEvent    = "add"
	DeleteEvent = "delete"
	KillEvent   = "kill"
)

Types

type Execute

type Execute interface {
	Execute(ctx *ExecuteContext) error
}

type ExecuteContext

type ExecuteContext struct {
	Job        *Job
	CancelCtx  context.Context
	CancelFunc context.CancelFunc
	IsLock     bool
}

type IStore

type IStore interface {
	Save(job *Job) error
	Delete(job *Job) error
	Lock(job *ExecuteContext) error
	UnLock(job *ExecuteContext) error
	Load() ([]*Job, error)
	Watch(WatchFunc)
}

IStore 存储

type Job

type Job struct {
	ExecuteKey string `json:"execute_key"` //执行器的key
	Key        string `json:"key"`         //脚本主键
	Data       []byte `json:"data"`        //脚本实体
	CronExpr   string `json:"cron_expr"`   //触发表达式
}

Job 脚本实体

func NewJob

func NewJob(execute, key string) *Job

NewJob 创建脚本

func (*Job) String

func (j *Job) String() string

func (*Job) Struct

func (j *Job) Struct(bean interface{})

func (*Job) WithCronTime

func (j *Job) WithCronTime(t time.Time)

func (*Job) WithData

func (j *Job) WithData(data []byte)

func (*Job) WithStructToData

func (j *Job) WithStructToData(data interface{})

type Manage

type Manage struct {
	Store IStore
	// contains filtered or unexported fields
}

Manage 脚本管理

func NewManage

func NewManage(store IStore) *Manage

func (*Manage) Delete

func (m *Manage) Delete(job *Job) error

Delete 删除

func (*Manage) Load

func (m *Manage) Load() ([]*Job, error)

Load 获取所有脚本

func (*Manage) On

func (m *Manage) On(key string, execute Execute)

func (*Manage) Run

func (m *Manage) Run()

func (*Manage) Save

func (m *Manage) Save(job *Job) error

Save 保存

type Memory

type Memory struct {
	// contains filtered or unexported fields
}

func NewMemory

func NewMemory() *Memory

func (*Memory) Delete

func (m *Memory) Delete(job *Job) error

func (*Memory) Load

func (m *Memory) Load() ([]*Job, error)

func (*Memory) Lock

func (m *Memory) Lock(ex *ExecuteContext) error

func (*Memory) Save

func (m *Memory) Save(job *Job) error

func (*Memory) Test

func (m *Memory) Test(job *Job)

func (*Memory) UnLock

func (m *Memory) UnLock(ex *ExecuteContext) error

func (*Memory) Watch

func (m *Memory) Watch(f WatchFunc)

type RedisStore

type RedisStore struct {
	// contains filtered or unexported fields
}

func NewRedisStore

func NewRedisStore(address string, options ...redis.DialOption) (*RedisStore, error)

func (*RedisStore) Delete

func (s *RedisStore) Delete(job *Job) (err error)

func (*RedisStore) Load

func (s *RedisStore) Load() ([]*Job, error)

func (*RedisStore) Lock

func (s *RedisStore) Lock(ex *ExecuteContext) error

func (*RedisStore) Save

func (s *RedisStore) Save(job *Job) error

func (*RedisStore) UnLock

func (s *RedisStore) UnLock(ex *ExecuteContext) error

func (*RedisStore) Watch

func (s *RedisStore) Watch(f WatchFunc)

type Scheduler

type Scheduler struct {
	Job     *Job
	EntryID cron.EntryID
	// contains filtered or unexported fields
}

Scheduler 执行计划

func (*Scheduler) Next

func (s *Scheduler) Next(t time.Time) time.Time

Next 计划的下一次执行时间

func (*Scheduler) Run

func (s *Scheduler) Run()

type WatchFunc

type WatchFunc func(event string, job *Job)