A persistent and flexible background jobs library for go.

Jobs

Development Status

Jobs is no longer being actively developed. I will still try my best to respond to issues and pull requests, but in general you should not expect much support. No new features will be added. Still, Jobs is reasonably well-tested, and it is probably fine to use it for low-traffic hobby sites. If you are looking for something for more serious, production use-cases, consider alternatives such as RabbitMQ.

Jobs follows semantic versioning but offers no guarantees of backwards compatibility until version 1.0.

About

Jobs is a persistent and flexible background jobs library for go.

Version Circle CI GoDoc

Jobs is powered by Redis and supports the following features:

  • A job can encapsulate any arbitrary functionality. A job can do anything which can be done in a go function.
  • A job can be one-off (only executed once) or recurring (scheduled to execute at a specific interval).
  • A job can be retried a specified number of times if it fails.
  • A job is persistent, with protections against power loss and other worst case scenarios. (See the Guarantees section below)
  • Work on jobs can be spread amongst any number of concurrent workers across any number of machines.
  • Provided it is persisted to disk, every job will be executed at least once, and in ideal network conditions will be executed exactly once. (See the Guarantees section below)
  • You can query the database to find out e.g. the number of jobs that are currently executing or how long a particular job took to execute.
  • Any job that permanently fails will have its error captured and stored.

Why is it Useful?

Jobs is intended to be used in web applications. It is useful for cases where you need to execute some long-running code, but you don't want your users to wait for the code to execute before rendering a response. A good example is sending a welcome email to your users after they sign up. You can use Jobs to schedule the email to be sent asynchronously, and render a response to your user without waiting for the email to be sent. You could use a goroutine to accomplish the same thing, but in the event of a server restart or power loss, the email might never be sent. Jobs guarantees that the email will be sent at some time and allows you to spread the work between different machines.

Installation

Jobs requires Go version >= 1.2. If you do not already have it, follow these instructions:

Jobs requires access to a Redis database. If you plan to have multiple worker pools spread out across different machines, they should all connect to the same Redis database. If you only want to run one worker pool, it is safe to install Redis locally and run it on the same machine. In either case, if you need to install Redis, follow these instructions:

  • Install Redis.
  • Follow the instructions in the section called Installing Redis more properly.
  • Make sure you understand how Redis Persistence works and have edited your config file to get your desired persistence. We recommend using both RDB and AOF and setting fsync to either "always" or "everysec".

After that, you can install Jobs like you would any other go package: go get github.com/albrow/jobs. If you want to update the package later, use go get -u github.com/albrow/jobs. Then you can import Jobs like you would any other go package by adding import github.com/albrow/jobs to your go source file.

Quickstart Guide

Connecting to Redis

You can configure the connection to Redis by editing Config.Db. Here are the options:

  • Address is the address of the redis database to connect to. Default is "localhost:6379".
  • Network is the type of network to use to connect to the redis database Default is "tcp".
  • Database is the redis database number to use for storing all data. Default is 0.
  • Password is a password to use for connecting to a redis database via the AUTH command. If empty, Jobs will not attempt to authenticate. Default is "" (an empty string).

You should edit Config.Db during program initialization, before running Pool.Start or scheduling any jobs. Here's an example of how to configure Jobs to use databse #10 and authenticate with the password "foobar":

func main() {
	// Configure database options at the start of your application
	jobs.Config.Db.Database = 10
	jobs.Config.Db.Password = "foobar"
}

Registering Job Types

Jobs must be organized into discrete types. Here's an example of how to register a job which sends a welcome email to users:

// We'll specify that we want the job to be retried 3 times before finally failing
welcomeEmailJobs, err := jobs.RegisterType("welcomeEmail", 3, func(user *User) error {
	msg := fmt.Sprintf("Hello, %s! Thanks for signing up for foo.com.", user.Name)
	if err := emails.Send(user.EmailAddress, msg); err != nil {
		// The returned error will be captured by a worker, which will then log the error
		// in the database and trigger up to 3 retries.
		return err
	}
})

The final argument to the RegisterType function is a HandlerFunc which will be executed when the job runs. HandlerFunc must be a function which accepts either zero or one arguments and returns an error.

Scheduling a Job

After registering a job type, you can schedule a job using the Schedule or ScheduleRecurring methods like so:

// The priority argument lets you choose how important the job is. Higher
// priority jobs will be executed first.
job, err := welcomeEmailJobs.Schedule(100, time.Now(), &User{EmailAddress: "[email protected]"})
if err != nil {
	// Handle err
}

You can use the Job object returned by Schedule or ScheduleRecurring to check on the status of the job or cancel it manually.

Starting and Configuring Worker Pools

You can schedule any number of worker pools across any number of machines, provided every machine agrees on the definition of the job types. If you want, you can start a worker pool on the same machines that are scheduling jobs, or you can have each worker pool running on a designated machine. Since each pool is assigned an id based on a unique hardware identifier, you must only run one worker pool per machine.

To create a new pool with the default configuration, just pass in nil:

pool, err := jobs.NewPool(nil)
if err != nil {
	// Handle err
}

You can also specify a different configuration by passing in *PoolConfig. Any zero values in the config you pass in will fallback to the default values. So here's how you could start a pool with 10 workers and a batch size of 10, while letting the other options remain the default.

pool, err := jobs.NewPool(&jobs.PoolConfig{
	NumWorkers: 10,
	BatchSize: 10,
})
if err != nil {
	// Handle err
}

After you have created a pool, you can start it with the Start method. Once started, the pool will continuously query the database for new jobs and delegate those jobs to workers. Any program that calls Pool.Start() should also wait for the workers to finish before exiting. You can do so by wrapping Close and Wait in a defer statement. Typical usage looks something like this:

func main() {
	pool, err := jobs.NewPool(nil)
	if err != nil {
		// Handle err
	}
	defer func() {
		pool.Close()
		if err := pool.Wait(); err != nil {
			// Handle err
		}
	}()
	if err := pool.Start(); err != nil {
		// Handle err
	}
}

You can also call Close and Wait at any time to manually stop the pool from executing new jobs. In this case, any jobs that are currently being executed will still finish.

Testing

To run the tests, make sure you have Redis running and accepting unix socket connections on the address /tmp/redis.sock. The tests will use database #14. WARNING: After each test is run, database #14 will be completely erased, so make sure you do not have any important data stored there.

To run the tests just run go test . If anything fails, please report an issue and describe what happened.

Contributing

See Contributing.md

Guarantees

Persistence

Since jobs is powered by Redis, there is a chance that you can lose data with the default Redis configuration. To get the best persistence guarantees, you should set Redis to use both AOF and RDB persistence modes and set fsync to "always". With these settings, Redis is more or less as persistent as a database like postgres. If want better performance and are okay with a slightly greater chance of losing data (i.e. jobs not executing), you can set fsync to "everysec".

Read more about Redis persistence.

Atomicity

Jobs is carefully written using Redis transactions and lua scripting so that all database changes are atomic. If Redis crashes in the middle of a transaction or script execution, it is possible that your AOF file can become corrupted. If this happens, Redis will refuse to start until the AOF file is fixed. It is relatively easy to fix the problem with the redis-check-aof tool, which will remove the partial transaction from the AOF file. In effect, this guarantees that modifications of the database are atomic, even in the event of a power loss or hard reset, with the caveat that you may need to use the redis-check-aof tool in the worst case scenario.

Read more about Redis transactions and scripts.

Job Execution

Jobs guarantees that a job will be executed at least once, provided it has been persisted on disk. (See the section on Persistence directly above). A job can only picked up by one pool at a time because a pool atomically pops (gets and immediately moves) the next available jobs from the database. A job can only be executed by one worker at a time because the jobs are delegated to workers via a shared channel. Each worker pool checks on the health of all the other pools when it starts. If a pool crashes or is otherwise disconnected, any jobs it had grabbed from the database that did not yet finish will be re-queued and picked up by a different pool.

This is in no way an exhaustive list, but here are some known examples of scenarios that may cause a job to be executed more than once:

  1. If there is a power failure or hard reset while a worker is in the middle of executing a job, the job may be stuck in a half-executed state. Since there is no way to know how much of the job was successfully completed, the job will be re-queued and picked up by a different pool, where it may be partially or fully executed more than once.
  2. If a pool becomes disconnected, it will be considered stale and its jobs will be re-queued and reclaimed by a different pool. However, if the stale pool is able to partly or fully execute jobs without a reliable internet connection, any jobs belonging to the stale pool might be executed more than once. You can increase the StaleTimeout parameter for a pool to make this scenario less likely.

License

Jobs is licensed under the MIT License. See the LICENSE file for more information.

Owner
Comments
  • job not exec with panic

    job not exec with panic

    image

    my code

    
    var sche scheduler
    
    type scheduler struct {
        tp           *jobs.Type
        golbalHeader map[string]string
    }
    
    func main() (err error) {
        tp, err := jobs.RegisterType(DEFAULT_JOB_TYPE, RETRY_TIMES, sche.getHandler())
        if err != nil {
            utils.Log().Error(err)
            return
        }
        sche.tp = tp
        pool, err := jobs.NewPool(nil)
        if err != nil {
            utils.Log().Error(err)
            return
        }
        defer func() {
            pool.Close()
            if err = pool.Wait(); err != nil {
                utils.Log().Error(err)
                return
            }
        }()
        if err = pool.Start(); err != nil {
            utils.Log().Error(err)
            return
        }
        return
    }
    
    ... 
    
    
    
    func (p *scheduler) getHandler() handler {
        return func(req *models.AddReq) error {
            utils.Log().Info("job start")
            post := gorequest.New().Post(req.Target)
            for k, v := range p.golbalHeader {
                post = post.Set(k, v)
            }
            for k, v := range req.Header {
                post = post.Set(k, v)
            }
            _, body, errs := post.Send(string(req.Body)).End()
            if len(errs) > 0 {
                utils.Log().Error(errs)
                return errs[0]
            }
            var rep response
            err := json.Unmarshal([]byte(body), &rep)
            if err != nil {
                utils.Log().Error(err)
                return err
            }
            if rep.Code != 0 {
                utils.Log().Error(rep.Code, rep.Message)
                return errors.New(rep.Message)
            }
            utils.Log().Info(rep)
            return nil
        }
    }
    
  • Error when calling Destroy() from within a job handler function

    Error when calling Destroy() from within a job handler function

    I've been seeing errors like the following lately. @albrow I was wondering if you could help me diagnose them.

    ERR Error running script (call to f_7be0eed842fab7944ce1393b76bf8f46826c6656): @user_script:20: us er_script:20: attempt to concatenate local 'Status' (a boolean value) I found the line it does this here

    I'm trying to destroy the job from within my handler function once it reaches a certain state. Is it not safe to do this?

    When I look up the job id I just get finished and time fields using hgetall.

    redis:6379[3]> hgetall jobs:bX8Vye9LGk80SDkbawj9qt3vvkpi
    1) "finished"
    2) "1435185638756948580"
    3) "time"
    4) "1435200038500419354"
    redis:6379[3]> 
    

    Is it supposed to leave the job in redis after you destroy it? From the destroy_script.lua docs it dosen't sound like it.

  • FindById() doesn't return error for missing jobs

    FindById() doesn't return error for missing jobs

    FindById() will return an empty job struct(with the id set to whatever you gave it) and nil if the id doesn't exist.

    Example

    j, err: = jobs.FindById("Hello world")
    if err != nil {
        panic(err)
    }
    log.Println("Found job:", j, err)
    

    @albrow It looks like transaction.exec() doesn't check for an empty set/list error.

  • Endless loop?

    Endless loop?

    What would be the best way / practical way to have an endless loop of the pool? From the current looks of it / examples, the pool will automatically close when there are no new jobs.

    I know that you can simply add:

    for {
    }
    

    To the end since the pools are running as goroutines.

    But shouldn't this behavior be by default? Or maybe I'm just too used to using Resque haha.

  • Use go generate to convert lua script files to strings

    Use go generate to convert lua script files to strings

    Over at albrow/zoom, it was brought to my attention that reading scripts from a file at runtime can cause problems for people using certain dependency managers, or people who want to release a binary executable. See PRs https://github.com/albrow/zoom/pull/9, https://github.com/albrow/zoom/pull/10, and https://github.com/albrow/zoom/pull/11.

    The solution I came up with was to use a small script (compatible with go generate) to read the contents from the .lua file and write them to a generated file as strings. I'm going to port a similar solution over to albrow/jobs. It will be a little bit more complicated because jobs uses templates for string constants in the lua scripts, but the general idea is the same. This is targeted for version 0.3.0 and will hopefully be released in the next couple days.

  • Passing params / string name of job.

    Passing params / string name of job.

    Similarly to Resque, you only have to pass in the name of the function ("HelloWorldJob") and a list of params { foo:"bar" }, etc.

    Is this possible with jobs? (or perhaps in future #14 implementation?)

    Rough example (more pseudo-code than Golang lol):

    job, err := Schedule("HelloWorldJob", 100, time.Now(), "{EmailAddress: "[email protected]"}")
    if err != nil {
        // Handle err
    }
    

    instead of

    job, err := welcomeEmailJobs.Schedule(100, time.Now(), &User{EmailAddress: "[email protected]"})
    if err != nil {
        // Handle err
    }
    

    That means that within whatever program (whether it is an http server or whatnot), I don't have to define all jobs to say a:

    var (
     sche *jobs.Type
    )
    

    everytime

  • After-job hook

    After-job hook

    This PR adds a function to the pool that gets executed each time a job is finished. Such function can be useful for monitoring.

    I thought about adding a full-fledged middleware chains first, but it would add a lot of complexity without too many benefits. This implementation gets a function which can accept a _Job via (_Pool).SetAfterFunc() and runs it each time a job is finished regardless of its status, passing the job to the function.

    Also, I moved the job.finished assgnment before the error checks (worker.go:85); this way the timings are saved even when the job returned an error.

  • Add a changelog

    Add a changelog

    I think it'd be nice if there was a changelog. This can either be done by hand or it can be automated if you force a syntax for commits ie changeType(component): msg. For an example of an automated changelog take a look at the angular-material project. Their changelog is generated by the commit messages and then I think they tweak any problems with it before a release. That might be a bit much for this project but it let's you know what can be done. Either way I think a basic changelog would be really nice.

    @albrow What do you think about this? I think it'd definitely make this easier to use in production.

  • Allow for multiple arguments to job handler functions

    Allow for multiple arguments to job handler functions

    Currently job handler functions can only accept one argument. This is not a priority for me, but it would not be too difficult to allow for multiple parameters.

  • gofmt

    gofmt

    Just ran

    gofmt -w .
    

    on the project root. That's all.

    https://blog.golang.org/go-fmt-your-code


    I made this PR with a project going on over at https://github.com/rotblauer/gofmt-att, and it's definitely a work in progress. So if I got something wrong, or this is annoying at all, please file an issue over there and we'll sort it out. On the other hand, if you think it's kind of funny like I do, please feel welcome to contribute or even fire up one of your own great automatic _gofmt_ers.

  • After-job hook

    After-job hook

    This PR adds a function to the pool that gets executed each time a job is finished. Such function can be useful for monitoring.

    I thought about adding a full-fledged middleware chains first, but it would add a lot of complexity without too many benefits. This implementation gets a function which can accept a Job via (Pool).SetAfterFunc() and runs it each time a job is finished regardless of its status, passing the job to the function.

    Also, I moved the job.finished assgnment before the error checks (worker.go:85); this way the timings are saved even when the job returned an error.

    (resubmitted, previous PR: #29)

  • Job status is changed to executing but job has not started execution

    Job status is changed to executing but job has not started execution

    Hi, I have scheduled a job at 9/15/2017, 1:58:36 PM IST epoch timestamp. The job got in executing status at this time and remained in executing state for next 5 mins without starting. It started at 9/15/2017, 2:03:08 PM IST. There were no other jobs to be executed as this was very first job. Please have a look at this, as this much delay will lead to problems in our application.

    Screen Shot of job details as job is hanged for execution screen shot 2017-09-15 at 2 03 53 pm

    Screen shot of job when it started screen shot 2017-09-15 at 2 04 13 pm

  • Support environmental prefix

    Support environmental prefix

    Problem: Current version of the package does not support environmental prefixes, hence if a common redis instance is being used between two environments (lets say alpha and beta) then there is no mechanism to ensure that consumers/workers are executing jobs from producers of that particular environment or some other. Proposed solution: One of the approach is to prefix the environment string to all the redis keys. Hence, there should be a feature to set prefix which varies as per the environment. Also, before implementing that at go layer, the hard coding of keys in the lua script should also be configurable in the sense that it should accept prefix as one of the arguments.

  • super slow

    super slow

    I benchmarked a few job processing libraries that use Redis as a backend: https://github.com/gocraft/work#benchmarks

    Most can do about 10,000 to 20,000 jobs/second. Using default options, the albrow/jobs library clocked in at 40 jobs/second. I was able to increase that by increasing the # of works and the batch size, but I wasn't sure what you think good values for those params are.

    Benchmark code: https://github.com/gocraft/work/blob/master/benches/bench_jobs/main.go

    I'd love it if you could review my benchmark to see if I made a glaring mistake, and also what your thoughts are about good params for batch size.

  • Application can't start due to error which message is 'jobs: In scanJob: Could not find Type with name = %s'

    Application can't start due to error which message is 'jobs: In scanJob: Could not find Type with name = %s'

    FYI, How to fix this problem?

    There has some job haven't execute while we restart the application. I think that may be redis callback application to run the job, but the memory hasn't the job object because of we restart. We only to clear the data that stored in redis to fix it, but i think this is not well.

    I hope you can understand what i say, sorry for my poor english.

    Best regards.

  • Error Handling?

    Error Handling?

    Is it possible to add a common error handler for all jobs that return an error so that I can say log my errors during development and send mails during production. Because right now in development I just return the error and don't know when and where the job failed.

Efficient and reliable background processing for Go

CurlyQ CurlyQ provides a simple, easy-to-use interface for performing background processing in Go. It supports scheduled jobs, job deduplication, and

Nov 11, 2022
Simple, efficient background processing for Golang backed by RabbitMQ and Redis
Simple, efficient background processing for Golang backed by RabbitMQ and Redis

Table of Contents How to Use Motivation Requirements Features Examples Setup Config Client/Server Task Worker/Task Hander Register The Handlers Send t

Nov 10, 2022
Run Jobs on a schedule, supports fixed interval, timely, and cron-expression timers; Instrument your processes and expose metrics for each job.

A simple process manager that allows you to specify a Schedule that execute a Job based on a Timer. Schedule manage the state of this job allowing you to start/stop/restart in concurrent safe way. Schedule also instrument this Job and gather metrics and optionally expose them via uber-go/tally scope.

Dec 8, 2022
Jenkins is a wonderful system for managing builds, and people love using its UI to configure jobs
Jenkins is a wonderful system for managing builds, and people love using its UI to configure jobs

Jenkins Job DSL Plugin Introduction Jenkins is a wonderful system for managing builds, and people love using its UI to configure jobs. Unfortunately,

Dec 20, 2022
gron, Cron Jobs in Go.

gron Gron provides a clear syntax for writing and deploying cron jobs. Goals Minimalist APIs for scheduling jobs. Thread safety. Customizable Job Type

Dec 20, 2022
YTask is an asynchronous task queue for handling distributed jobs in golang
YTask is an asynchronous task queue for handling distributed jobs in golang

YTask is an asynchronous task queue for handling distributed jobs in golang

Dec 24, 2022
A way of scheduling volcano jobs

JobFlow 背景 volcano Volcano是CNCF 下首个也是唯一的基于Kubernetes的容器批量计算平台,主要用于高性能计算场景。 它提供了Kubernetes目前缺 少的一套机制,这些机制通常是机器学习大数据应用、科学计算、 特效渲染等多种高性能工作负载所需的。 现状:当前vol

Oct 12, 2022
Scheduler: Go jobs execution system

Scheduler Go jobs execution system. Inspired by CI/CD and Unity task scheduler.

Jul 1, 2022
Executes jobs in separate GO routines. Provides Timeout, StartTime controls. Provides Cancel all running job before new job is run.

jobExecutor Library to execute jobs in GO routines. Provides for Job Timeout/Deadline (MaxDuration()) Job Start WallClock control (When()) Add a job b

Jan 10, 2022
Graceful shutdown with repeating "cron" jobs (running at a regular interval) in Go

Graceful shutdown with repeating "cron" jobs (running at a regular interval) in Go Illustrates how to implement the following in Go: run functions ("j

May 30, 2022
Marshmallow provides a flexible and performant JSON unmarshalling in Go. It specializes in dealing with unstructured struct - when some fields are known and some aren't, with zero performance overhead nor extra coding needed.
Marshmallow provides a flexible and performant JSON unmarshalling in Go. It specializes in dealing with unstructured struct - when some fields are known and some aren't, with zero performance overhead nor extra coding needed.

Marshmallow Marshmallow package provides a simple API to perform flexible and performant JSON unmarshalling in Go. Marshmallow specializes in dealing

Dec 26, 2022
A simple Cron library for go that can execute closures or functions at varying intervals, from once a second to once a year on a specific date and time. Primarily for web applications and long running daemons.

Cron.go This is a simple library to handle scheduled tasks. Tasks can be run in a minimum delay of once a second--for which Cron isn't actually design

Dec 17, 2022
clockwork - Simple and intuitive job scheduling library in Go.
clockwork - Simple and intuitive job scheduling library in Go.

clockwork A simple and intuitive scheduling library in Go. Inspired by python's schedule and ruby's clockwork libraries. Example use package main imp

Jul 27, 2022
Chrono is a scheduler library that lets you run your task and code periodically
Chrono is a scheduler library that lets you run your task and code periodically

Chrono is a scheduler library that lets you run your tasks and code periodically. It provides different scheduling functionalities to make it easier t

Dec 26, 2022
A zero-dependencies and lightweight go library for job scheduling

A zero-dependencies and lightweight go library for job scheduling.

Aug 3, 2022
Simple, zero-dependency scheduling library for Go

go-quartz Simple, zero-dependency scheduling library for Go. About Inspired by the Quartz Java scheduler. Library building blocks Job interface. Any t

Dec 30, 2022
a cron library for go

cron Cron V3 has been released! To download the specific tagged release, run: go get github.com/robfig/cron/[email protected] Import it in your program as: im

Dec 25, 2022
go-sche is a golang library that lets you schedule your task to be executed later.

go-sche is a golang library that lets you schedule your task to be executed later.

Dec 24, 2022
Scheduler - Scheduler package is a zero-dependency scheduling library for Go

Scheduler Scheduler package is a zero-dependency scheduling library for Go Insta

Jan 14, 2022