README
¶
Queue
Queue is a Golang library for spawning and managing a Goroutine pool, Allowing 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.
- Support Redis Pub/Sub as backend.
Queue Scenario
In Single Container or VM
Multile Container with Queue service like NSQ, NATs or Redis
Installation
Install the stable version:
go get github.com/golang-queue/queue
Install the latest verison:
go get github.com/golang-queue/queue@master
Usage
Basic usage of Pool (use Task function)
By calling QueueTask() method, it schedules the task executed by worker (goroutines) in the Pool.
package main
import (
"context"
"fmt"
"time"
"github.com/golang-queue/queue"
)
func main() {
taskN := 100
rets := make(chan string, taskN)
// initial queue pool
q := queue.NewPool(5)
// shutdown the service and notify all the worker
// wait all jobs are complete.
defer q.Release()
// assign tasks in queue
for i := 0; i < taskN; i++ {
go func(i int) {
if err := q.QueueTask(func(ctx context.Context) error {
rets <- fmt.Sprintf("Hi Gopher, handle the job: %02d", +i)
return nil
}); err != nil {
panic(err)
}
}(i)
}
// wait until all tasks done
for i := 0; i < taskN; i++ {
fmt.Println("message:", <-rets)
time.Sleep(20 * time.Millisecond)
}
}
Basic usage of Pool (use message queue)
Define the new message struct and implement the Bytes() func to encode message. Give the WithFn func
to handle the message from Queue.
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"time"
"github.com/golang-queue/queue"
)
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)
// initial queue pool
q := queue.NewPool(5, queue.WithFn(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
}))
// shutdown the service and notify all the worker
// wait all jobs are complete.
defer q.Release()
// assign tasks in queue
for i := 0; i < taskN; i++ {
go func(i int) {
if err := q.Queue(&job{
Name: "Gopher",
Message: fmt.Sprintf("handle the job: %d", i+1),
}); err != nil {
log.Println(err)
}
}(i)
}
// wait until all tasks done
for i := 0; i < taskN; i++ {
fmt.Println("message:", <-rets)
time.Sleep(50 * time.Millisecond)
}
}
Using NSQ as Queue
See the NSQ documentation.
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"time"
"github.com/golang-queue/nsq"
"github.com/golang-queue/queue"
)
type job struct {
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 := 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
}),
)
// define the queue
q := queue.NewPool(
5,
queue.WithWorker(w),
)
// assign tasks in queue
for i := 0; i < taskN; i++ {
go func(i int) {
q.Queue(&job{
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.Release()
}
Using NATs as Queue
See the NATs documentation
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"time"
"github.com/golang-queue/nats"
"github.com/golang-queue/queue"
)
type job struct {
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 := 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
}),
)
// define the queue
q, err := queue.NewQueue(
queue.WithWorkerCount(10),
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{
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.Release()
}
Using Redis(Pub/Sub) as Queue
See the redis documentation
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"time"
"github.com/golang-queue/queue"
"github.com/golang-queue/redisdb"
)
type job struct {
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 := redisdb.NewWorker(
redisdb.WithAddr("127.0.0.1:6379"),
redisdb.WithChannel("foobar"),
redisdb.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
}),
)
// define the queue
q, err := queue.NewQueue(
queue.WithWorkerCount(10),
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{
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.Release()
}
Documentation
¶
Index ¶
- Variables
- type Consumer
- type Job
- type Logger
- type Metric
- type Option
- type Options
- type Queue
- 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) Release()
- 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 ¶
Constants ¶
This section is empty.
Variables ¶
View Source
var ErrMissingWorker = errors.New("missing worker module")
ErrMissingWorker missing define worker
View Source
var ErrQueueShutdown = errors.New("queue has been closed and released")
ErrQueueShutdown the queue is released and closed.
Functions ¶
This section is empty.
Types ¶
type Consumer ¶ added in v0.0.7
type Consumer struct {
// contains filtered or unexported fields
}
Worker for simple queue using channel
func (*Consumer) BusyWorkers ¶ added in v0.0.8
func (*Consumer) Queue ¶ added in v0.0.7
func (s *Consumer) Queue(job QueuedMessage) error
Queue send notification to queue
type Job ¶
type Job struct {
Task TaskFunc `json:"-"`
Timeout time.Duration `json:"timeout"`
Body []byte `json:"body"`
}
Job with Timeout
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")
Output:
type Metric ¶ added in v0.0.10
type Metric interface {
IncBusyWorker()
DecBusyWorker()
BusyWorkers() uint64
}
Metric interface
type Option ¶
type Option func(*Options)
Option for queue system
func WithFn ¶ added in v0.0.7
func WithFn(fn func(context.Context, QueuedMessage) error) Option
WithFn set custom job function
func WithQueueSize ¶ added in v0.0.7
WithQueueSize set worker count
func WithTimeOut ¶ added in v0.0.9
WithTimeOut set custom timeout
type Options ¶ added in v0.0.7
type Options struct {
// contains filtered or unexported fields
}
func NewOptions ¶ added in v0.0.7
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
A Queue is a message queue.
func (*Queue) QueueTaskWithTimeout ¶
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
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
// BusyWorkers return count of busy worker currently
BusyWorkers() uint64
}
Worker interface
Source Files
Click to show internal directories.
Click to hide internal directories.