jobkicker is A Golang in-process cron task scheduler that kicks (executes) them once in specified time or periodically.
- Kick (execute) jobs to run after some time once or periodically.
- Kick (execute) jobs to run at certain time.
- Ability to cancel any job with it's id.
- Logs execution and canceling jobs with the flexibility to save the logs to any thing implements
io.Writerinterface like a file or even to implement a writer to write the logs to something like elastic search and pass it to the scheduler - flexibility to format the logs as you wish by passing the Formatter interface from sirupsen/logrus
- Uses language built-in time.Time to reduce design complexity by not using something like cron time format.
- Run multiply scheduled jobs concurrently at the same time.
- sirupsen/logrus for logging.
- google/uuid for generating job's id.
The main type in the package which is the scheduler that kicks new jobs to run and cancels them and holds the JobQueue and the Logger.
An interface type which KickerTimer and KickerTicker implement it which they are just wrappers for time.Timer and time.Ticker.
Job struct that holds the function with it's arguments and the timer to execute it.
JobQueue is just holding PendingJobs which is just a map for the pending jobs to be excuted map[string]*Job and DoneJobs which is a map for executed jobs with it's last execution time and a read/write mutex to lock this two maps when accessed.
Note: periodically executed functions stays in PendingJobs map after execution (unless canceled) and DoneJobs keeps track of the time of it's last execution
Returns a new JobKicker (scheduler) and takes:
loggerOutputwhich any type implements the interfaceio.Writerto write the logs to, and if nil passed it will write toos.Stderr. the interfaceio.Writeris:
type Writer interface {
Write(p []byte) (n int, err error)
}loggerFormatterwhich any type implements the interfacelogrus.Formatterinterface, and if nil passed it uselogrus.TextFormatterby default, you can try to pass&logrus.JSONFormatter{}to format the logs as json or pass your custom formatter that implements:
type Formatter interface {
Format(*Entry) ([]byte, error)
}func (jobKicker *JobKicker) KickOnceAfter(delay time.Time, fn interface{}, args ...interface{}) (jobID string)
Runs a function once after a given delay, the delay is a time.Time type with all fields zero expect the time to runs it after, as if you want to run it after 3 hours and 30 minutes create a new time with time.Date(year ,month ,day ,hour ,min ,sec,nsec,loc) with all fields parameters equal zero expect hour = 3 and min = 30, and the second parameter is the function to run and the rest of the parameters are the function arguments if any.
example:
import (
"time"
"github.com/MohabMohamed/jobkicker"
)
func main() {
task := func() {
println("jobkicker is awesome")
}
jk := jobkicker.NewScheduler(nil, nil)
// time.Date(year ,month ,day ,hour ,min ,sec,nsec,loc)
// every field equals zeo expect seconds equals 3
delay := time.Date(0, 0, 0, 0, 0, 3, 0, time.UTC)
jk.KickOnceAfter(delay, task)
time.Sleep(4 * time.Second)
}Output:
jobkicker is awesome
INFO[0003] job with id [5e1b8baa-5133-483b-9188-8179ecc8aea4] executed in 2022-02-08 22:43:08.701094256 +0200 EET m=+3.002681837
func (jobKicker *JobKicker) KickOnceAt(runAt time.Time, fn interface{}, args ...interface{}) (jobID string)
Runs a function once at a certain time, for example if you want to run a function at 1 march 2022 13:30 am you should create a time.Time with this certain time like time.Date(2022 ,3 ,1 ,13 ,30 ,0 , 0, time.UTC) and the function will run at that time. the second parameter is the function and the rest are the function arguments.
example:
package main
import (
"fmt"
"time"
"github.com/MohabMohamed/jobkicker"
)
func main() {
task := func() {
println("jobkicker is awesome")
}
jk := jobkicker.NewScheduler(nil, nil)
// time.Date(year ,month ,day ,hour ,min ,sec,nsec,loc)
// KickOnceAt takes the date to run the task in
// in this example I run it with date equals now + 3 seconds
fmt.Println(time.Now())
runAt := time.Now().Add(3 * time.Second)
jk.KickOnceAt(runAt, task)
time.Sleep(4 * time.Second)
}Output:
2022-02-09 12:39:26.645499827 +0200 EET m=+0.000422605
jobkicker is awesome
INFO[0003] job with id [0477e575-22a1-48a2-851f-017b5aeb9ea4] executed in 2022-02-09 12:39:29.646510482 +0200 EET m=+3.001433319
func (jobKicker *JobKicker) KickPeriodicallyEvery(delay time.Time, fn interface{}, args ...interface{}) (jobID string)
Runs the function every some specified time intervals it takes the delay like KickOnceAfter so if you pass time.Time with 3 seconds it will run the function every 3 seconds, and the second parameter is the function and the rest are the function arguments.
example:
example:
package main
import (
"time"
"github.com/MohabMohamed/jobkicker"
)
func main() {
task := func() {
println("jobkicker is awesome")
}
jk := jobkicker.NewScheduler(nil, nil)
// time.Date(year ,month ,day ,hour ,min ,sec,nsec,loc)
// every field equals zeo expect seconds equals 3
// as KickPeriodicallyEvery takes the delay between every execution
delay := time.Date(0, 0, 0, 0, 0, 3, 0, time.UTC)
jk.KickPeriodicallyEvery(delay, task)
time.Sleep(10 * time.Second)
}Output:
jobkicker is awesome
INFO[0003] job with id [d8ae6a99-1d2b-434f-8a9c-db77a4e3e844] executed in 2022-02-09 12:43:54.758287996 +0200 EET m=+3.001531441
jobkicker is awesome
INFO[0009] job with id [d8ae6a99-1d2b-434f-8a9c-db77a4e3e844] executed in 2022-02-09 12:43:57.760335343 +0200 EET m=+6.003578840
jobkicker is awesome
INFO[0011] job with id [d8ae6a99-1d2b-434f-8a9c-db77a4e3e844] executed in 2022-02-09 12:44:00.758331522 +0200 EET m=+9.001575009
Cancels the scheduling of a job, it takes it's id and return error if the job of type run once and already ran or if it's not already scheduled (maybe wrong id given)
example:
example:
package main
import (
"fmt"
"time"
"github.com/MohabMohamed/jobkicker"
)
func main() {
task := func() {
println("jobkicker is awesome")
}
jk := jobkicker.NewScheduler(nil, nil)
// time.Date(year ,month ,day ,hour ,min ,sec,nsec,loc)
// KickOnceAt takes the date to run the task in
// in this example I run it with date equals now + 3 seconds
fmt.Println(time.Now())
runAt := time.Now().Add(3 * time.Second)
jk.KickOnceAt(runAt, task)
time.Sleep(4 * time.Second)
}Output:
2022-02-09 12:39:26.645499827 +0200 EET m=+0.000422605
jobkicker is awesome
INFO[0003] job with id [0477e575-22a1-48a2-851f-017b5aeb9ea4] executed in 2022-02-09 12:39:29.646510482 +0200 EET m=+3.001433319
func (jobKicker *JobKicker) KickPeriodicallyEvery(delay time.Time, fn interface{}, args ...interface{}) (jobID string)
Runs the function every some specified time intervals it takes the delay like KickOnceAfter so if you pass time.Time with 3 seconds it will run the function every 3 seconds, and the second parameter is the function and the rest are the function arguments.
example:
example:
package main
import (
"fmt"
"time"
"github.com/MohabMohamed/jobkicker"
)
func main() {
task := func() {
println("jobkicker is awesome")
}
jk := jobkicker.NewScheduler(nil, nil)
// time.Date(year ,month ,day ,hour ,min ,sec,nsec,loc)
// every field equals zeo expect seconds equals 3
// as KickPeriodicallyEvery takes the delay between every execution
delay := time.Date(0, 0, 0, 0, 0, 3, 0, time.UTC)
jobId := jk.KickPeriodicallyEvery(delay, task)
// sleep for 4 seconds to let it run once before cancelling
time.Sleep(4 * time.Second)
err := jk.CancelJob(jobId)
if err != nil {
fmt.Printf("error while cancelling a job: %s", err.Error())
}
time.Sleep(10 * time.Second)
}Output:
jobkicker is awesome
INFO[0003] job with id [e7e0d6ab-6aa5-4b52-9f60-3c74e5585b15] executed in 2022-02-09 12:49:14.102550679 +0200 EET m=+3.001345112
INFO[0016] job with id [e7e0d6ab-6aa5-4b52-9f60-3c74e5585b15] cancelled successfully in 2022-02-09 12:49:15.101968589 +0200 EET m=+4.000763046
you can use the apis in any combination to kick the jobs, for example if you want to kick a job every year at 1 march 2022 13:30 am you should create a time.Time with this certain time like time.Date(2022 ,3 ,1 ,13 ,30 ,0 , 0, time.UTC) and delay with 1 year and pass KickPeriodicallyEvery with arguments the delay and the function to kick and it's arguments to KickOnceAt the date specified like:
package main
import (
"fmt"
"time"
"github.com/MohabMohamed/jobkicker"
)
func main() {
task := func(name string) {
fmt.Printf("jobkicker is awesome. don't you agree,%s?\n", name)
}
jk := jobkicker.NewScheduler(nil, nil)
date := time.Date(2022, 3, 1, 13, 30, 0, 0, time.UTC)
delay := time.Date(1, 0, 0, 0, 0, 0, 0, time.UTC)
jk.KickPeriodicallyEvery(delay, jk.KickOnceAt, date, task, "Mohab")
// block the main goroutine, could be server.listen() or any thing
for true {
}
}Get creative using jobkicker, and keep kicking these tasks :D
-
Made the job execution (timers and context) self contained in the job to make it easier to cancel.
-
Used Read/Write mutex instead of regular mutex as the only write operations made to the JobQueue when remove the job from pendeningJobs (Run once job and executed or a canceled job) and when executing job adding the last execution time to the done jobs, the rest is read operations.
-
Use regular map with RWMutex instead of sync.map as I have 2 maps so with 2 sync.map both of them will lock and unlock and both of them need to be locked at the same time so lock and unlock a mutex and lock and unlock another one will be performance costly more than using 1 mutex.
-
Using already built-in
time.Timeinstead of rolling of my solution to handle delay and time as every go developer is familier with them so it would easier for the user.
check contribution guide and the Reference
I'm considering maybe to add the ability to consist the tasks execution in redis as option as if the client code that using jobkicker got down and up again could reschedule the tasks that already scheduled.
Maybe adding the ability to schedule the tasks in distributed environment as if a task ran on a machine it shouldn't run from another one.