ontime

README

ontime

介绍

用于需要大量在规定时间内判定某个任务是完成 正常情况下可以对某个任务开启一个线程，但是在任务量多的情况下后台会开启很多无效线程。 或者可以使用mq，比如借助死信队列，但是额外增加了依赖。 所以此包的目的是为了减少无效线程的数量和对三方软件的依赖。 实现方式参考了cron库

使用场景示例

比如我们在客服中心系统中，我们对未接电话会发送一条短信询问用户是否需要 回电，如果用户在规定时间内(比如十分钟）回复了需要回电， 则客服需要回电给客服，超时或者回复了不需要回电则不做任何处理。 常见的操作是把当前任务加入数据库，然后轮询数据库中不断检测任务是否完成。 这个时候我们使用此库的话。示例代码如下：

package main

import (
	"gitee.com/hellochuang/ontime"
	"gitee.com/hellochuang/xlog"
	"math/rand"
	"time"
)

// 停止ontime任务后执行的东西，会自动在调用stop之后执行
func afterDo(entries []*ontime.Entry) {
	for _, entry := range entries {
		xlog.Info(entry.EntryId, entry.ExpireAt)
	}
}

// 检测短信回复内容
func checkSms() int {
	// 0-1
	return rand.Intn(2)
}

func main() {
	c := ontime.New(ontime.WithAfter(afterDo))
	c.Run()
	var answered = false
	if !answered {
		c.Add(&ontime.Entry{
			EntryId:  0,
			ExpireAt: time.Now().Add(time.Minute * 10),
			Job: func() error {
				if checkSms() == 0 {
					xlog.Info("无需回电")
				} else {
					xlog.Info("需要回电..做点儿什么吧")
				}
				return nil
			},
		})
	}
        // 会等待所有正在执行的任务完成后才返回，并自动调用注入的afterRun函数，对未完成的entries进行处理，比如可以存入数据库或者别的处理
        c.Stop()
        
	select {}
}

参与贡献

特技

Documentation

Index

• type Entry
• type EntryChain
  • func (e EntryChain) Len() int
  • func (e EntryChain) Less(i, j int) bool
  • func (e EntryChain) Swap(i, j int)
• type FileStore
  • func (fs FileStore) Read() ([]*Entry, error)
  • func (fs FileStore) Write(data interface{}) error
• type Job
• type Logger
  • func PrintfLogger(l interface{ ... }) Logger
• type OnTime
  • func New(options ...Option) *OnTime
  • func (onTime *OnTime) Add(job Job, expireAt time.Time) int64
  • func (onTime *OnTime) Remove(entryId int64)
  • func (onTime *OnTime) Start()
  • func (onTime *OnTime) Stop() context.Context
• type Option
  • func WithAfter(f func([]*Entry)) Option
  • func WithBefore(f func()) Option
  • func WithLogger(logger Logger) Option
• type Store
  • func NewFileStore() Store

Types

type Entry

type Entry struct {
	EntryId  int64
	ExpireAt time.Time
	Job      Job
	//unique tag, if not empty, we can use it to remove duplicate entries,not only by EntryId
	UniqueTag string
}

type EntryChain

type EntryChain []*Entry

func (EntryChain) Len

func (e EntryChain) Len() int

func (EntryChain) Less

func (e EntryChain) Less(i, j int) bool

func (EntryChain) Swap

func (e EntryChain) Swap(i, j int)

type FileStore

type FileStore struct{}

func (FileStore) Read

func (fs FileStore) Read() ([]*Entry, error)

func (FileStore) Write

func (fs FileStore) Write(data interface{}) error

type Job

type Job func() error

type Logger

type Logger interface {
	Info(v ...interface{})
	Error(v ...interface{})
}

var DefaultLogger Logger = PrintfLogger(log.New(os.Stdout, "on_time: ", log.LstdFlags))

func PrintfLogger

func PrintfLogger(l interface{ Print(...interface{}) }) Logger

type OnTime

type OnTime struct {
	WaitGroup sync.WaitGroup
	// contains filtered or unexported fields
}

func New

func New(options ...Option) *OnTime

func (*OnTime) Add

func (onTime *OnTime) Add(job Job, expireAt time.Time) int64

Add add one job function with expire

func (*OnTime) Remove

func (onTime *OnTime) Remove(entryId int64)

func (*OnTime) Start

func (onTime *OnTime) Start()

func (*OnTime) Stop

func (onTime *OnTime) Stop() context.Context

type Option

type Option func(o *OnTime)

func WithAfter

func WithAfter(f func([]*Entry)) Option

func WithBefore

func WithBefore(f func()) Option

func WithLogger

func WithLogger(logger Logger) Option

type Store

type Store interface {
	Write(data interface{}) error
	Read() ([]*Entry, error)
}

func NewFileStore

func NewFileStore() Store