nsq

README

NSQ

NSQ as backend with Queue package (A realtime distributed messaging platform)

Setup

start the NSQ lookupd

nsqlookupd

start the NSQ server

nsqd --lookupd-tcp-address=localhost:4160

start the NSQ admin dashboard

nsqadmin --lookupd-http-address localhost:4161

Testing

go test -v ./...

Example

package main

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

  "github.com/golang-queue/nsq"
  "github.com/golang-queue/queue"
  "github.com/golang-queue/queue/core"
)

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 core.TaskMessage) error {
      var v *job
      if err := json.Unmarshal(m.Payload(), &v); err != nil {
        return err
      }
      rets <- v.Message
      return nil
    }),
  )

  // define the queue
  q := queue.NewPool(
    10,
    queue.WithWorker(w),
  )
  defer q.Release()

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

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

Documentation

Index

• type Option
  • func WithAddr(addr string) Option
  • func WithChannel(channel string) Option
  • func WithLogLevel(lvl nsq.LogLevel) Option
  • func WithLogger(l queue.Logger) Option
  • func WithMaxInFlight(num int) Option
  • func WithRunFunc(fn func(context.Context, core.TaskMessage) error) Option
  • func WithTopic(topic string) Option
• type OptionFunc
  • func (f OptionFunc) Apply(option *Options)
• type Options
• type Worker
  • func NewWorker(opts ...Option) *Worker
  • func (w *Worker) Queue(job core.TaskMessage) error
  • func (w *Worker) Request() (core.TaskMessage, error)
  • func (w *Worker) Run(ctx context.Context, task core.TaskMessage) error
  • func (w *Worker) Shutdown() error
  • func (w *Worker) Stats() *nsq.ConsumerStats

Types

type Option

type Option interface {
	Apply(*Options)
}

An Option configures a mutex.

func WithAddr

func WithAddr(addr string) Option

WithAddr setup the addr of NSQ

func WithChannel

func WithChannel(channel string) Option

WithChannel setup the channel of NSQ

func WithLogLevel

func WithLogLevel(lvl nsq.LogLevel) Option

WithLogLevel set custom nsq log level

func WithLogger

func WithLogger(l queue.Logger) Option

WithLogger set custom logger

func WithMaxInFlight

func WithMaxInFlight(num int) Option

WithMaxInFlight Maximum number of messages to allow in flight (concurrency knob)

func WithRunFunc

func WithRunFunc(fn func(context.Context, core.TaskMessage) error) Option

WithRunFunc setup the run func of queue

func WithTopic

func WithTopic(topic string) Option

WithTopic setup the topic of NSQ

type OptionFunc

type OptionFunc func(*Options)

OptionFunc is a function that configures a queue.

func (OptionFunc) Apply

func (f OptionFunc) Apply(option *Options)

Apply calls f(option)

type Options

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

type Worker

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

Worker for NSQ

func NewWorker

func NewWorker(opts ...Option) *Worker

NewWorker for struc

func (*Worker) Queue

func (w *Worker) Queue(job core.TaskMessage) error

Queue send notification to queue

func (*Worker) Request

func (w *Worker) Request() (core.TaskMessage, error)

Request fetch new task from queue

func (*Worker) Run

func (w *Worker) Run(ctx context.Context, task core.TaskMessage) error

Run start the worker

func (*Worker) Shutdown

func (w *Worker) Shutdown() error

Shutdown worker

func (*Worker) Stats

func (w *Worker) Stats() *nsq.ConsumerStats

Stats retrieves the current connection and message statistics for a Consumer