workers

README

Sidekiq compatible background workers in golang.

• reliable queueing for all queues using brpoplpush
• handles retries
• support custom middleware
• customize concurrency per queue
• responds to Unix signals to safely wait for jobs to finish before exiting.
• provides stats on what jobs are currently running
• redis sentinel support
• well tested

Example usage:

package main

import (
	"github.com/digitalocean/go-workers2"
)

func myJob(message *workers.Msg) error {
  // do something with your message
  // message.Jid()
  // message.Args() is a wrapper around go-simplejson (http://godoc.org/github.com/bitly/go-simplejson)
  return nil
}

func myMiddleware(queue string, mgr *Manager, next JobFunc) JobFunc {
  return func(message *workers.Msg) (err error) {
    // do something before each message is processed
    err = next()
    // do something after each message is processed
    return
  }
}

func main() {
  // Create a manager, which manages workers
  manager, err := workers.NewManager(Options{
    // location of redis instance
    ServerAddr: "localhost:6379",
    // instance of the database
    Database:   0,
    // number of connections to keep open with redis
    PoolSize:   30,
    // unique process id for this instance of workers (for proper recovery of inprogress jobs on crash)
    ProcessID:  "1",
  })

  // create a middleware chain with the default middlewares, and append myMiddleware
  mids := workers.DefaultMiddlewares().Append(myMiddleware)

  // pull messages from "myqueue" with concurrency of 10
  // this worker will not run myMiddleware, but will run the default middlewares
  manager.AddWorker("myqueue", myJob, 10)

  // pull messages from "myqueue2" with concurrency of 20
  // this worker will run the default middlewares and myMiddleware
  manager.AddWorker("myqueue2", myJob, 20, mids...)

  // pull messages from "myqueue3" with concurrency of 20
  // this worker will only run myMiddleware
  manager.AddWorker("myqueue3", myJob, 20, myMiddleware)

  // Create a producer to enqueue messages
  producer, err := workers.NewProducer(Options{
    // location of redis instance
    ServerAddr: "localhost:6379",
    // instance of the database
    Database:   0,
    // number of connections to keep open with redis
    PoolSize:   30,
    // unique process id for this instance of workers (for proper recovery of inprogress jobs on crash)
    ProcessID:  "1",
  })
  // Alternatively, if you already have a manager and want to enqueue
  // to the same place:
  producer := manager.Producer()

  // Add a job to a queue
  producer.Enqueue("myqueue3", "Add", []int{1, 2})

  // Add a job to a queue with retry
  producer.EnqueueWithOptions("myqueue3", "Add", []int{1, 2}, workers.EnqueueOptions{Retry: true})

  // stats will be available at http://localhost:8080/stats
  go workers.StartStatsServer(8080)

  // Blocks until process is told to exit via unix signal
  manager.Run()
}

Development sponsored by DigitalOcean. Code forked from github/jrallison/go-workers. Initial development sponsored by Customer.io.

Documentation

Index

• Constants
• func StartStatsServer(port int)
• type Args
  • func (d Args) Equals(other interface{}) bool
  • func (d Args) ToJson() string
• type EnqueueData
• type EnqueueOptions
• type Fetcher
• type JobFunc
  • func LogMiddleware(queue string, mgr *Manager, next JobFunc) JobFunc
  • func NopMiddleware(queue string, mgr *Manager, final JobFunc) JobFunc
  • func RetryMiddleware(queue string, mgr *Manager, next JobFunc) JobFunc
  • func StatsMiddleware(queue string, mgr *Manager, next JobFunc) JobFunc
• type JobStatus
• type Manager
  • func NewManager(options Options) (*Manager, error)
  • func (m *Manager) AddBeforeStartHooks(hooks ...func())
  • func (m *Manager) AddDuringDrainHooks(hooks ...func())
  • func (m *Manager) AddWorker(queue string, concurrency int, job JobFunc, mids ...MiddlewareFunc)
  • func (m *Manager) GetRedisClient() *redis.Client
  • func (m *Manager) GetStats() (Stats, error)
  • func (m *Manager) Producer() *Producer
  • func (m *Manager) RetryQueue() string
  • func (m *Manager) Run()
  • func (m *Manager) Stop()
• type MiddlewareFunc
• type Middlewares
  • func DefaultMiddlewares() Middlewares
  • func NewMiddlewares(mids ...MiddlewareFunc) Middlewares
  • func (m Middlewares) Append(mid MiddlewareFunc) Middlewares
  • func (m Middlewares) Prepend(mid MiddlewareFunc) Middlewares
• type Msg
  • func NewMsg(content string) (*Msg, error)
  • func (m *Msg) Args() *Args
  • func (d Msg) Equals(other interface{}) bool
  • func (m *Msg) Jid() string
  • func (m *Msg) OriginalJson() string
  • func (d Msg) ToJson() string
• type Options
• type Producer
  • func NewProducer(options Options) (*Producer, error)
  • func (p *Producer) Enqueue(queue, class string, args interface{}) (string, error)
  • func (p *Producer) EnqueueAt(queue, class string, at time.Time, args interface{}) (string, error)
  • func (p *Producer) EnqueueIn(queue, class string, in float64, args interface{}) (string, error)
  • func (p *Producer) EnqueueWithOptions(queue, class string, args interface{}, opts EnqueueOptions) (string, error)
  • func (p *Producer) GetRedisClient() *redis.Client
• type Stats
• type WorkersLogger

Constants

const (
	DefaultRetryMax = 25
	RetryTimeFormat = "2006-01-02 15:04:05 MST"
)

const (
	NanoSecondPrecision = 1000000000.0
)

Functions

func StartStatsServer

func StartStatsServer(port int)

Types

type Args

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

func (Args) Equals

func (d Args) Equals(other interface{}) bool

func (Args) ToJson

func (d Args) ToJson() string

type EnqueueData

type EnqueueData struct {
	Queue      string      `json:"queue,omitempty"`
	Class      string      `json:"class"`
	Args       interface{} `json:"args"`
	Jid        string      `json:"jid"`
	EnqueuedAt float64     `json:"enqueued_at"`
	EnqueueOptions
}

type EnqueueOptions

type EnqueueOptions struct {
	RetryCount int     `json:"retry_count,omitempty"`
	Retry      bool    `json:"retry,omitempty"`
	At         float64 `json:"at,omitempty"`
}

type Fetcher

type Fetcher interface {
	Queue() string
	Fetch()
	Acknowledge(*Msg)
	Ready() chan bool
	Messages() chan *Msg
	Close()
	Closed() bool
}

type JobFunc

type JobFunc func(message *Msg) error

func LogMiddleware

func LogMiddleware(queue string, mgr *Manager, next JobFunc) JobFunc

func NopMiddleware

func NopMiddleware(queue string, mgr *Manager, final JobFunc) JobFunc

NopMiddleware does nothing

func RetryMiddleware

func RetryMiddleware(queue string, mgr *Manager, next JobFunc) JobFunc

func StatsMiddleware

func StatsMiddleware(queue string, mgr *Manager, next JobFunc) JobFunc

type JobStatus

type JobStatus struct {
	Message   *Msg  `json:"message"`
	StartedAt int64 `json:"started_at"`
}

type Manager

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

func NewManager

func NewManager(options Options) (*Manager, error)

func (*Manager) AddBeforeStartHooks

func (m *Manager) AddBeforeStartHooks(hooks ...func())

func (*Manager) AddDuringDrainHooks

func (m *Manager) AddDuringDrainHooks(hooks ...func())

func (*Manager) AddWorker

func (m *Manager) AddWorker(queue string, concurrency int, job JobFunc, mids ...MiddlewareFunc)

func (*Manager) GetRedisClient

func (m *Manager) GetRedisClient() *redis.Client

func (*Manager) GetStats

func (m *Manager) GetStats() (Stats, error)

func (*Manager) Producer

func (m *Manager) Producer() *Producer

func (*Manager) RetryQueue

func (m *Manager) RetryQueue() string

func (*Manager) Run

func (m *Manager) Run()

Run starts all workers under this Manager and blocks until they exit.

func (*Manager) Stop

func (m *Manager) Stop()

Stop all workers under this Manager and returns immediately.

type MiddlewareFunc

type MiddlewareFunc func(queue string, m *Manager, next JobFunc) JobFunc

type Middlewares

type Middlewares []MiddlewareFunc

func DefaultMiddlewares

func DefaultMiddlewares() Middlewares

func NewMiddlewares

func NewMiddlewares(mids ...MiddlewareFunc) Middlewares

func (Middlewares) Append

func (m Middlewares) Append(mid MiddlewareFunc) Middlewares

func (Middlewares) Prepend

func (m Middlewares) Prepend(mid MiddlewareFunc) Middlewares

type Msg

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

func NewMsg

func NewMsg(content string) (*Msg, error)

func (*Msg) Args

func (m *Msg) Args() *Args

func (Msg) Equals

func (d Msg) Equals(other interface{}) bool

func (*Msg) Jid

func (m *Msg) Jid() string

func (*Msg) OriginalJson

func (m *Msg) OriginalJson() string

func (Msg) ToJson

func (d Msg) ToJson() string

type Options

type Options struct {
	ProcessID    string
	Namespace    string
	PollInterval int
	Database     int
	Password     string
	PoolSize     int

	// Provide one of ServerAddr or (SentinelAddrs + RedisMasterName)
	ServerAddr      string
	SentinelAddrs   string
	RedisMasterName string

	// Optional display name used when displaying manager stats
	ManagerDisplayName string
	// contains filtered or unexported fields
}

type Producer

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

func NewProducer

func NewProducer(options Options) (*Producer, error)

func (*Producer) Enqueue

func (p *Producer) Enqueue(queue, class string, args interface{}) (string, error)

func (*Producer) EnqueueAt

func (p *Producer) EnqueueAt(queue, class string, at time.Time, args interface{}) (string, error)

func (*Producer) EnqueueIn

func (p *Producer) EnqueueIn(queue, class string, in float64, args interface{}) (string, error)

func (*Producer) EnqueueWithOptions

func (p *Producer) EnqueueWithOptions(queue, class string, args interface{}, opts EnqueueOptions) (string, error)

func (*Producer) GetRedisClient

func (p *Producer) GetRedisClient() *redis.Client

type Stats

type Stats struct {
	Name      string                 `json:"manager_name"`
	Processed int64                  `json:"processed"`
	Failed    int64                  `json:"failed"`
	Jobs      map[string][]JobStatus `json:"jobs"`
	Enqueued  map[string]int64       `json:"enqueued"`
	Retries   int64                  `json:"retries"`
}

type WorkersLogger

type WorkersLogger interface {
	Println(...interface{})
	Printf(string, ...interface{})
}

var Logger WorkersLogger = log.New(os.Stdout, "workers: ", log.Ldate|log.Lmicroseconds)