queue

README

Queue

Queue is a Golang library for spawning and managing a Goroutine pool, Alloowing you to create multiple worker according to limit CPU number of machine.

Features

• Support buffered channel queue.
• Support NSQ (A realtime distributed messaging platform) as backend.
• Support NATS (Connective Technology for Adaptive Edge & Distributed Systems) as backend.

Installation

Install the stable version:

go get github.com/appleboy/queue

Install the latest verison:

go get github.com/appleboy/queue@master

Usage

The first step to create a new job as QueueMessage interface:

type job struct {
  Message string
}

func (j *job) Bytes() []byte {
  b, err := json.Marshal(j)
  if err != nil {
    panic(err)
  }
  return b
}

The second step to create the new worker, use the buffered channel as an example, you can use the stop channel to terminate the job immediately after shutdown the queue service if need.

// define the worker
w := simple.NewWorker(
  simple.WithQueueNum(taskN),
  simple.WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
    v, ok := m.(*job)
    if !ok {
      if err := json.Unmarshal(m.Bytes(), &v); err != nil {
        return err
      }
    }

    rets <- v.Message
    return nil
  }),
)

or use the NSQ as backend, see the worker example:

// define the worker
w := nsq.NewWorker(
  nsq.WithAddr("127.0.0.1:4150"),
  nsq.WithTopic("example"),
  nsq.WithChannel("foobar"),
  // concurrent job number
  nsq.WithMaxInFlight(10),
  nsq.WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
    v, ok := m.(*job)
    if !ok {
      if err := json.Unmarshal(m.Bytes(), &v); err != nil {
        return err
      }
    }

    rets <- v.Message
    return nil
  }),
)

or use the NATS as backend, see the worker example:

w := nats.NewWorker(
  nats.WithAddr("127.0.0.1:4222"),
  nats.WithSubj("example"),
  nats.WithQueue("foobar"),
  nats.WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
    v, ok := m.(*job)
    if !ok {
      if err := json.Unmarshal(m.Bytes(), &v); err != nil {
        return err
      }
    }

    rets <- v.Message
    return nil
  }),
)

The third step to create a queue and initialize multiple workers, receive all job messages:

// define the queue
q, err := queue.NewQueue(
  queue.WithWorkerCount(5),
  queue.WithWorker(w),
)
if err != nil {
  log.Fatal(err)
}

// start the five worker
q.Start()

// assign tasks in queue
for i := 0; i < taskN; i++ {
  go func(i int) {
    q.Queue(&job{
      Name:    "foobar",
      Message: fmt.Sprintf("handle the job: %d", i+1),
    })
  }(i)
}

// wait until all tasks done
for i := 0; i < taskN; i++ {
  fmt.Println("message:", <-rets)
  time.Sleep(50 * time.Millisecond)
}

// shutdown the service and notify all the worker
q.Shutdown()
// wait all jobs are complete.
q.Wait()

Full example code as below or try it in playground.

package main

import (
  "context"
  "encoding/json"
  "fmt"
  "log"
  "time"

  "github.com/appleboy/queue"
  "github.com/appleboy/queue/simple"
)

type job struct {
  Name    string
  Message string
}

func (j *job) Bytes() []byte {
  b, err := json.Marshal(j)
  if err != nil {
    panic(err)
  }
  return b
}

func main() {
  taskN := 100
  rets := make(chan string, taskN)

  // define the worker
  w := simple.NewWorker(
    simple.WithQueueNum(taskN),
    simple.WithRunFunc(func(ctx context.Context, m queue.QueuedMessage) error {
      v, ok := m.(*job)
      if !ok {
        if err := json.Unmarshal(m.Bytes(), &v); err != nil {
          return err
        }
      }

      rets <- "Hi, " + v.Name + ", " + v.Message
      return nil
    }),
  )

  // define the queue
  q, err := queue.NewQueue(
    queue.WithWorkerCount(5),
    queue.WithWorker(w),
  )
  if err != nil {
    log.Fatal(err)
  }

  // start the five worker
  q.Start()

  // assign tasks in queue
  for i := 0; i < taskN; i++ {
    go func(i int) {
      q.Queue(&job{
        Name:    "foobar",
        Message: fmt.Sprintf("handle the job: %d", i+1),
      })
    }(i)
  }

  // wait until all tasks done
  for i := 0; i < taskN; i++ {
    fmt.Println("message:", <-rets)
    time.Sleep(50 * time.Millisecond)
  }

  // shutdown the service and notify all the worker
  q.Shutdown()
  // wait all jobs are complete.
  q.Wait()
}

Documentation

Index

• Variables
• type Job
  • func (j Job) Bytes() []byte
  • func (j Job) Encode() []byte
• type Logger
  • func NewEmptyLogger() Logger
  • func NewLogger() Logger
• type Option
  • func WithLogger(l Logger) Option
  • func WithWorker(w Worker) Option
  • func WithWorkerCount(num int) Option
• type Queue
  • func NewQueue(opts ...Option) (*Queue, error)
  • func (q *Queue) Capacity() int
  • func (q *Queue) Queue(job QueuedMessage) error
  • func (q *Queue) QueueTask(task TaskFunc) error
  • func (q *Queue) QueueTaskWithTimeout(timeout time.Duration, task TaskFunc) error
  • func (q *Queue) QueueWithTimeout(timeout time.Duration, job QueuedMessage) error
  • func (q *Queue) Shutdown()
  • func (q *Queue) Start()
  • func (q *Queue) Usage() int
  • func (q *Queue) Wait()
  • func (q *Queue) Workers() int
• type QueuedMessage
• type TaskFunc
• type Worker

Examples

• NewEmptyLogger

Variables

var ErrMissingWorker = errors.New("missing worker module")

ErrMissingWorker missing define worker

var ErrQueueShutdown = errors.New("queue has been closed")

ErrQueueShutdown close the queue.

Types

type Job

type Job struct {
	Task    TaskFunc      `json:"-"`
	Timeout time.Duration `json:"timeout"`
	Body    []byte        `json:"body"`
}

Job with Timeout

func (Job) Bytes

func (j Job) Bytes() []byte

Bytes get string body

func (Job) Encode

func (j Job) Encode() []byte

type Logger

type Logger interface {
	Infof(format string, args ...interface{})
	Errorf(format string, args ...interface{})
	Fatalf(format string, args ...interface{})
	Info(args ...interface{})
	Error(args ...interface{})
	Fatal(args ...interface{})
}

Logger interface is used throughout gorush

func NewEmptyLogger

func NewEmptyLogger() Logger

NewEmptyLogger for simple logger.

Example

l := NewEmptyLogger()
l.Info("test")
l.Infof("test")
l.Error("test")
l.Errorf("test")
l.Fatal("test")
l.Fatalf("test")

func NewLogger

func NewLogger() Logger

NewLogger for simple logger.

type Option

type Option func(*Queue)

Option for queue system

func WithLogger

func WithLogger(l Logger) Option

WithLogger set custom logger

func WithWorker

func WithWorker(w Worker) Option

WithWorker set custom worker

func WithWorkerCount

func WithWorkerCount(num int) Option

WithWorkerCount set worker count

type Queue

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

A Queue is a message queue.

func NewQueue

func NewQueue(opts ...Option) (*Queue, error)

NewQueue returns a Queue.

func (*Queue) Capacity

func (q *Queue) Capacity() int

Capacity for queue max size

func (*Queue) Queue

func (q *Queue) Queue(job QueuedMessage) error

Queue to queue all job

func (*Queue) QueueTask

func (q *Queue) QueueTask(task TaskFunc) error

QueueTask to queue job task

func (*Queue) QueueTaskWithTimeout

func (q *Queue) QueueTaskWithTimeout(timeout time.Duration, task TaskFunc) error

QueueTaskWithTimeout to queue job task with timeout

func (*Queue) QueueWithTimeout

func (q *Queue) QueueWithTimeout(timeout time.Duration, job QueuedMessage) error

Queue to queue all job

func (*Queue) Shutdown

func (q *Queue) Shutdown()

Shutdown stops all queues.

func (*Queue) Start

func (q *Queue) Start()

Start to enable all worker

func (*Queue) Usage

func (q *Queue) Usage() int

Usage for count of queue usage

func (*Queue) Wait

func (q *Queue) Wait()

Wait all process

func (*Queue) Workers

func (q *Queue) Workers() int

Workers returns the numbers of workers has been created.

type QueuedMessage

type QueuedMessage interface {
	Bytes() []byte
}

QueuedMessage ...

type TaskFunc

type TaskFunc func(context.Context) error

TaskFunc is the task function

type Worker

type Worker interface {
	// BeforeRun is called before starting the worker
	BeforeRun() error
	// Run is called to start the worker
	Run() error
	// BeforeRun is called after starting the worker
	AfterRun() error
	// Shutdown is called if stop all worker
	Shutdown() error
	// Queue to send message in Queue (single channel, NSQ or AWS SQS)
	Queue(job QueuedMessage) error
	// Capacity queue capacity = cap(channel name)
	Capacity() int
	// Usage is how many message in queue
	Usage() int
}

Worker interface