workers

package module
v0.0.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 26, 2022 License: MIT Imports: 24 Imported by: 0

README

Build Status GoDoc

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 (
  "fmt"

  workers "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 *workers.Manager, next workers.JobFunc) workers.JobFunc {
  return func(message *workers.Msg) (err error) {
    // do something before each message is processed
    err = next(message)
    // do something after each message is processed
    return
  }
}

func main() {
  // Create a manager, which manages workers
  manager, err := workers.NewManager(workers.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",
  })

  if err != nil {
    fmt.Println(err)
  }

  // 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", 10, myJob)

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

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

  // If you already have a manager and want to enqueue
  // to the same place:
  producer := manager.Producer()

  // Alternatively, if you want to 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",
  // })

  // 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.StartAPIServer(8080)

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

When running the above code example, it will produce the following output at localhost:8080/stats:

[
  {
    "manager_name": "",
    "processed": 5,
    "failed": 57,
    "jobs": {
      "myqueue": null,
      "myqueue2": null,
      "myqueue3": null
    },
    "enqueued": {
      "myqueue": 0,
      "myqueue2": 0,
      "myqueue3": 0
    },
    "retry_count": 4
  }
]

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

Documentation

Index

Constants

View Source 
const (
	// DefaultRetryMax is default for max number of retries for a job
	DefaultRetryMax = 25

	// RetryTimeFormat is default for retry time format
	RetryTimeFormat = "2006-01-02 15:04:05 MST"
)
View Source 
const (
	// NanoSecondPrecision is a constant for the number of nanoseconds in a second
	NanoSecondPrecision = 1000000000.0
)

Variables

View Source 
var Logger = log.New(os.Stdout, "go-workers2: ", log.Ldate|log.Lmicroseconds)

Logger is the default go-workers2 logger, only used here in this file. TODO: remove this

Functions

func ConfigureAPIServer 

func ConfigureAPIServer(options APIOptions)

ConfigureAPIServer allows global API server configuration with the given options

func RegisterAPIEndpoints 

func RegisterAPIEndpoints(mux *http.ServeMux)

RegisterAPIEndpoints sets up API server endpoints

func StartAPIServer 

func StartAPIServer(port int)

StartAPIServer starts the API server

func StopAPIServer 

func StopAPIServer()

StopAPIServer stops the API server

Types

type APIOptions 

type APIOptions struct {
	Logger *log.Logger
	Mux    *http.ServeMux
}

APIOptions contains the set of configuration options for the global api

type Args 

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

Args is the set of parameters for a message

func (Args) Equals 

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

func (Args) ToJson 

func (d Args) ToJson() string

ToJson return data in JSON format th message

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
}

EnqueueData stores data and configuration for new work

type EnqueueOptions 

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

EnqueueOptions stores configuration for new work

type Fetcher 

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

Fetcher is an interface for managing work messages

type JobFunc 

type JobFunc func(message *Msg) error

JobFunc is a message processor

func LogMiddleware 

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

LogMiddleware is the default logging middleware

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

RetryMiddleware middleware that allows retries for jobs failures

func StatsMiddleware 

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

StatsMiddleware middleware to collect stats on processed messages

type JobStatus 

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

JobStatus contains the status and data for active jobs of a manager

type Manager 

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

Manager coordinates work, workers, and signaling needed for job processing

func NewManager 

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

NewManager creates a new manager with provide options

func NewManagerWithRedisClient 

func NewManagerWithRedisClient(options Options, client *redis.Client) (*Manager, error)

NewManagerWithRedisClient creates a new manager with provide options and pre-configured Redis client

func (*Manager) AddBeforeStartHooks 

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

AddBeforeStartHooks adds functions to be executed before the manager starts

func (*Manager) AddDuringDrainHooks 

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

AddDuringDrainHooks adds function to be execute during a drain operation

func (*Manager) AddRetriesExhaustedHandlers 

func (m *Manager) AddRetriesExhaustedHandlers(handlers ...RetriesExhaustedFunc)

AddRetriesExhaustedHandlers adds function(s) to be executed when retries are exhausted for a job.

func (*Manager) AddWorker 

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

AddWorker adds a new job processing worker

func (*Manager) GetRedisClient 

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

GetRedisClient returns the Redis client used by the manager

func (*Manager) GetRetries 

func (m *Manager) GetRetries(page uint64, pageSize int64, match string) (Retries, error)

GetRetries returns the set of retry jobs for the manager

func (*Manager) GetStats 

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

GetStats returns the set of stats for the manager

func (*Manager) Producer 

func (m *Manager) Producer() *Producer

Producer creates a new work producer with configuration identical to the manager

func (*Manager) Run 

func (m *Manager) Run()

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

func (*Manager) SetRetriesExhaustedHandlers 

func (m *Manager) SetRetriesExhaustedHandlers(handlers ...RetriesExhaustedFunc)

SetRetriesExhaustedHandlers sets function(s) that will be sequentially executed when retries are exhausted for a job.

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

MiddlewareFunc is an extra function on the processing pipeline

type Middlewares 

type Middlewares []MiddlewareFunc

Middlewares contains the lists of all configured middleware functions

func DefaultMiddlewares 

func DefaultMiddlewares() Middlewares

DefaultMiddlewares creates the default middleware pipeline

func NewMiddlewares 

func NewMiddlewares(mids ...MiddlewareFunc) Middlewares

NewMiddlewares creates the processing pipeline given the list of middleware funcs

func (Middlewares) Append 

func (m Middlewares) Append(mid MiddlewareFunc) Middlewares

Append adds middleware to the end of the processing pipeline

func (Middlewares) Prepend 

func (m Middlewares) Prepend(mid MiddlewareFunc) Middlewares

Prepend adds middleware to the front of the processing pipeline

type Msg 

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

Msg is the struct for job data (parameters and metadata)

func NewMsg 

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

NewMsg returns a new message

func (*Msg) Args 

func (m *Msg) Args() *Args

Args returns arguments attribute of a message

func (*Msg) Class 

func (m *Msg) Class() string

Class returns class attribute of a message

func (Msg) Equals 

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

func (*Msg) Jid 

func (m *Msg) Jid() string

Jid returns job id attribute of a message

func (*Msg) OriginalJson 

func (m *Msg) OriginalJson() string

OriginalJson returns the original JSON message

func (Msg) ToJson 

func (d Msg) ToJson() string

ToJson return data in JSON format th message

type Options 

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

	// Provide one of ServerAddr or (SentinelAddrs + RedisMasterName)
	ServerAddr      string
	SentinelAddrs   string
	RedisMasterName string
	RedisTLSConfig  *tls.Config

	// Optional display name used when displaying manager stats
	ManagerDisplayName string

	// Log
	Logger *log.Logger
	// contains filtered or unexported fields
}

Options contains the set of configuration options for a manager and/or producer

type Producer 

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

Producer is used to enqueue new work

func NewProducer 

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

NewProducer creates a new producer with the given options

func NewProducerWithRedisClient 

func NewProducerWithRedisClient(options Options, client *redis.Client) (*Producer, error)

NewProducerWithRedisClient creates a new producer with the given options and Redis client

func (*Producer) Enqueue 

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

Enqueue enqueues new work for immediate processing

func (*Producer) EnqueueAt 

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

EnqueueAt enqueues new work for processing at a specific time

func (*Producer) EnqueueIn 

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

EnqueueIn enqueues new work for delayed processing

func (*Producer) EnqueueWithOptions 

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

EnqueueWithOptions enqueues new work for processing with the given options

func (*Producer) GetRedisClient 

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

GetRedisClient returns the Redis client used by the producer Deprecated: the Redis client is an internal implementation and access will be removed

type Retries 

type Retries struct {
	TotalRetryCount int64  `json:"total_retry_count"`
	RetryJobs       []*Msg `json:"retry_jobs"`
}

Retries stores retry information

type RetriesExhaustedFunc 

type RetriesExhaustedFunc func(queue string, message *Msg, err error)

RetriesExhaustedFunc gets executed when retry attempts have been exhausted.

type RetryOptions 

type RetryOptions struct {
	Exp      int `json:"exp"`
	MinDelay int `json:"min_delay"`
	MaxDelay int `json:"max_delay"`
	MaxRand  int `json:"max_rand"`
}

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"`
	RetryCount int64                  `json:"retry_count"`
}

Stats containts current stats for a manager

Source Files

View all Source files

Directories

Path Synopsis
cmd
gwctl

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.