README
¶
redis
Redis as backend for Queue package
Setup
start the redis server
redis-server
start the redis cluster, see the config
# server 01
mkdir server01 && cd server 01 && redis-server redis.conf --port 6379
# server 02
mkdir server02 && cd server 02 && redis-server redis.conf --port 6380
Example
For single server
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 := queue.NewPool(
5,
queue.WithWorker(w),
)
// 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(50 * time.Millisecond)
}
// shutdown the service and notify all the worker
q.Release()
}
Testing
go test -v ./...
Documentation
¶
Index ¶
- type Option
- func WithAddr(addr string) Option
- func WithChannel(channel string) Option
- func WithChannelSize(size int) Option
- func WithCluster(enable bool) Option
- func WithConnectionString(connectionString string) Option
- func WithDB(db int) Option
- func WithLogger(l queue.Logger) Option
- func WithMetric(m queue.Metric) Option
- func WithPassword(passwd string) Option
- func WithRunFunc(fn func(context.Context, queue.QueuedMessage) error) Option
- type Worker
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*options)
Option for queue system
func WithConnectionString ¶
WithConnectionString redis connection string
func WithMetric ¶ added in v0.0.6
WithMetric set custom Metric
func WithRunFunc ¶
WithRunFunc setup the run func of queue
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker for Redis
func (*Worker) BusyWorkers ¶ added in v0.0.3
BusyWorkers return count of busy workers currently.
Source Files
Click to show internal directories.
Click to hide internal directories.