README
¶
Jobs
A persistent and flexible background jobs library for go.
Version: 0.0.1
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.
- Jobs can be executed by any number of concurrent workers accross any number of machines, and each job will only be executed once.
- 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.
Quickstart Guide
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.RegisterJobType("welcomeEmail", 3, func(user *User) {
msg := fmt.Sprintf("Hello, %s! Thanks for signing up for foo.com.", user.Name)
if err := emails.Send(user.EmailAddress, msg); err != nil {
// Panics will be captured by a worker, triggering up to 3 retries
panic(err)
}
})
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: "foo@example.com"})
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 accross 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. It is technically safe to start multiple pools on a single machine, but typically you should only start one pool per machine.
To create a new pool with the default configuration, just pass in nil:
pool := jobs.NewPool(nil)
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 := jobs.NewPool(&jobs.PoolConfig{
NumWorkers: 10,
BatchSize: 10,
})
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 := jobs.NewPool(nil)
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.
License
Jobs is licensed under the MIT License. See the LICENSE file for more information.
Documentation
¶
Overview ¶
package jobs is a persistent and flexible background jobs library.
Version: 0.0.1
Jobs 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.
- Jobs can be executed by any number of concurrent workers accross any number of machines, and each job will only be executed once.
- 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.
More Information ¶
Visit https://github.com/albrow/jobs for a Quickstart Guide, code examples, and more information.
Index ¶
- Variables
- type ErrorNameAlreadyRegistered
- type Job
- func (j *Job) Cancel() error
- func (j *Job) Destroy() error
- func (j *Job) Duration() time.Duration
- func (j *Job) Enqueue() error
- func (j *Job) Error() error
- func (j *Job) Finished() time.Time
- func (j *Job) Id() string
- func (j *Job) Refresh() error
- func (j *Job) Started() time.Time
- func (j *Job) Status() JobStatus
- type JobStatus
- type JobType
- type Pool
- type PoolConfig
Constants ¶
This section is empty.
Variables ¶
var Config = configType{
Db: databaseConfig{
Address: "localhost:6379",
Network: "tcp",
Database: 0,
},
}
Config is where all configuration variables are stored. You may modify Config directly in order to change config variables, and should only do so at the start of your program.
var DefaultPoolConfig = &PoolConfig{ NumWorkers: runtime.GOMAXPROCS(0), BatchSize: runtime.GOMAXPROCS(0), MinWait: 200 * time.Millisecond, StaleTimeout: 30 * time.Second, }
DefaultPoolConfig is the default config for pools. You can override any values by passing in a *PoolConfig to NewPool. Any zero values in PoolConfig will be interpreted as the default.
Functions ¶
This section is empty.
Types ¶
type ErrorNameAlreadyRegistered ¶
type ErrorNameAlreadyRegistered struct {
// contains filtered or unexported fields
}
ErrorNameAlreadyRegistered is returned whenever RegisterJobType is called with a name that has already been registered.
func (ErrorNameAlreadyRegistered) Error ¶
func (e ErrorNameAlreadyRegistered) Error() string
Error satisfies the error interface.
type Job ¶
type Job struct {
// contains filtered or unexported fields
}
Job represents a discrete piece of work to be done by a worker.
func (*Job) Cancel ¶
Cancel cancels the job, but does not remove it from the database. It will be added to a list of cancelled jobs. If you wish to remove it from the database, use the Destroy method.
func (*Job) Destroy ¶
Destroy removes all traces of the job from the database. If the job is currently being executed by a worker, the worker may still finish the job.
func (*Job) Duration ¶
Duration returns how long the job took to execute with nanosecond precision. I.e. the difference between j.Finished() and j.Started(). It returns a duration of zero if the job has not finished yet.
func (*Job) Enqueue ¶
Enqueue adds the job to the queue and sets its status to StatusQueued. Queued jobs will be completed by workers in order of priority.
func (*Job) Error ¶
Error returns the last error that arose during execution of the job. It is only non-nil if the job has failed at some point.
func (*Job) Finished ¶
Finished returns the time that the job finished executing (in local time with nanosecond precision) or the zero time if the job has not finished executing yet.
func (*Job) Id ¶
Id returns the unique identifier used for the job. If the job has not yet been saved to the database, it may return an empty string.
func (*Job) Refresh ¶
Refresh mutates the job by setting its fields to the most recent data found in the database. It returns an error if there was a problem connecting to the database or if the job was destroyed.
type JobStatus ¶
type JobStatus string
JobStatus represents the different statuses a job can have.
const ( // StatusSaved is the status of any job that has been saved into the database but not yet queued StatusSaved JobStatus = "saved" // StatusQueued is the status of any job that has been queued for execution but not yet selected StatusQueued JobStatus = "queued" // StatusExecuting is the status of any job that has been selected for execution and is being delegated // to some worker and any job that is currently being executed by some worker. StatusExecuting JobStatus = "executing" // StatusFinished is the status of any job that has been successfully executed. StatusFinished JobStatus = "finished" // StatusFailed is the status of any job that failed to execute and for which there are no remaining retries. StatusFailed JobStatus = "failed" // StatusCancelled is the status of any job that was manually cancelled. StatusCancelled JobStatus = "cancelled" // StatusDestroyed is the status of any job that has been destroyed, i.e. completely removed // from the database. StatusDestroyed JobStatus = "destroyed" )
type JobType ¶
type JobType struct {
// contains filtered or unexported fields
}
JobType represents a type of job that can be executed by workers
func RegisterJobType ¶
RegisterJobType registers a new type of job that can be executed by workers. name should be a unique string identifier for the job. retries is the number of times this type of job should be retried if it fails. handler is a function that a worker will call in order to execute the job. handler should be a function which accepts either 0 or 1 arguments of any type, corresponding to the data for a job of this type. All jobs of this type must have data with the same type as the first argument to handler, or nil if the handler accepts no arguments.
func (*JobType) Schedule ¶
Schedule schedules a on-off job of the given type with the given parameters. Jobs with a higher priority will be executed first. The job will not be executed until after time. data is the data associated with this particular job and should have the same type as the first argument to the handler for this JobType.
func (*JobType) ScheduleRecurring ¶
func (jt *JobType) ScheduleRecurring(priority int, time time.Time, freq time.Duration, data interface{}) (*Job, error)
ScheduleRecurring schedules a recurring job of the given type with the given parameters. Jobs with a higher priority will be executed first. The job will not be executed until after time. After time, the job will be executed with a frequency specified by freq. data is the data associated with this particular job and should have the same type as the first argument to the handler for this JobType. Every recurring execution of the job will use the same data.
type Pool ¶
type Pool struct {
// RWMutex is only used during testing when we need to
// change some of the fields for the pool after it was started.
// NOTE: currently only used in one test (TestStalePoolsArePurged)
// and might be removed if we refactor later.
sync.RWMutex
// contains filtered or unexported fields
}
Pool is a pool of workers. Pool will query the database for queued jobs and delegate those jobs to some number of workers. It will do this continuously until the main program exits or you call Pool.Close().
func NewPool ¶
func NewPool(config *PoolConfig) *Pool
NewPool creates and returns a new pool with the given configuration. You can pass in nil to use the default values. Otherwise, any zero values in config will be interpreted as the default value.
func (*Pool) Close ¶
func (p *Pool) Close()
Close closes the worker pool and prevents it from delegating any new jobs. However, any jobs that are currently being executed will still be executed. Close returns immediately. If you want to wait until all workers are done executing their current jobs, use the Wait method.
func (*Pool) Start ¶
Start starts the worker pool. This means the pool will initialize workers, continuously query the database for queued jobs, and delegate those jobs to the workers.
func (*Pool) Wait ¶
Wait will return when all workers are done executing their jobs. Wait can only possibly return after you have called Close. To prevent errors due to partially-executed jobs, any go program which starts a worker pool should call Wait (and Close before that if needed) before exiting.
type PoolConfig ¶
type PoolConfig struct {
// NumWorkers is the number of workers to run
// Each worker will run inside its own goroutine
// and execute jobs asynchronously. Default is
// runtime.GOMAXPROCS.
NumWorkers int
// BatchSize is the number of jobs to send through
// the jobs channel at once. Increasing BatchSize means
// the worker pool will query the database less frequently,
// so you would get higher performance. However this comes
// at the cost that jobs with lower priority may sometimes be
// executed before jobs with higher priority, because the jobs
// with higher priority were not ready yet the last time the pool
// queried the database. Decreasing BatchSize means more
// frequent queries to the database and lower performance, but
// greater likelihood of executing jobs in perfect order with regards
// to priority. Setting BatchSize to 1 gaurantees that higher priority
// jobs are always executed first as soon as they are ready. Default is
// runtime.GOMAXPROCS.
BatchSize int
// MinWait is the minimum amount of time the pool will wait before checking
// the database for queued jobs. The pool may take longer to query the database
// if the jobs channel is blocking (i.e. if no workers are ready to execute new
// jobs). Default is 200ms.
MinWait time.Duration
// StaleTimeout is the amount of time to wait for a pool to reply to a ping request
// before considering it stale. Stale pools will be purged and if they have any
// corresponding jobs in the executing set, those jobs will be requeued. Default
// is 30 seconds.
StaleTimeout time.Duration
}
PoolConfig is a set of configuration options for pools. Setting any value to the zero value will be interpretted as the default.
Source Files