README
¶
redis
Redis Pub/Sub as backend for Queue package
Setup
start the redis server
redis-server
start the redis cluster, see the config
# server 01
mkdir server01 && cd server01 && redis-server redis.conf --port 6379
# server 02
mkdir server02 && cd server02 && 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.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(
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() Option
- func WithConnectionString(connectionString string) Option
- func WithDB(db int) Option
- func WithDebug() Option
- func WithLogger(l queue.Logger) Option
- func WithMasterName(masterName string) Option
- func WithPassword(passwd string) Option
- func WithRunFunc(fn func(context.Context, core.TaskMessage) error) Option
- func WithSentinel() Option
- func WithSkipTLSVerify() Option
- func WithTLS() Option
- func WithUsername(username string) 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 WithMasterName ¶ added in v0.1.0
WithMasterName sentinel master name
func WithRunFunc ¶
WithRunFunc setup the run func of queue
func WithSkipTLSVerify ¶ added in v0.2.0
func WithSkipTLSVerify() Option
WithSkipTLSVerify returns an Option that configures the TLS settings to skip verification of the server's certificate. This is useful for connecting to servers with self-signed certificates or when certificate verification is not required. Use this option with caution as it makes the connection susceptible to man-in-the-middle attacks.
func WithTLS ¶ added in v0.2.0
func WithTLS() Option
WithTLS returns an Option that configures the use of TLS for the connection. It sets the minimum TLS version to TLS 1.2.
func WithUsername ¶ added in v0.2.0
WithUsername redis username
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker for Redis
func NewWorker ¶
NewWorker creates a new Worker instance with the provided options. It initializes a Redis client based on the options and establishes a connection to the Redis server. The Worker is responsible for subscribing to a Redis channel and receiving messages from it. It returns the created Worker instance.
func (*Worker) Queue ¶
func (w *Worker) Queue(job core.TaskMessage) error
Queue send notification to queue
func (*Worker) Request ¶ added in v0.0.8
func (w *Worker) Request() (core.TaskMessage, error)
Request a new task
Source Files