redisOrderedQueue

redis-ordered-queue-go

Priority queue with message-group based partitioning and equal attention guarantee for each message group based on Redis

What is it?

Redis-based ordered queue with support for message priority.

Quality metrics

Queue guarantees

Low latency delivery

Based on traffic and amount of consumers, it is possible to reach very low delivery latency. I observed 2-15 millisecond latencies with thousands of consumers, message groups and messages.

Equal attention to all message groups

Message groups are cycled through, each consumer looking at the next available message group in the buffer.

No conflicts between message groups

Message groups are locked for processing before a consumer can process messages associated with that message group. This ensures that when a consumer is processing messages in a message group, no other consumers can see messages in the same message group.

The drawback is that if you only have one message group in your set-up, only one consumer can be active at each moment.

At least once delivery

Message redelivery attempts will take place until it's acknowledged by the consumer.

High performance, low memory footprint

The amount of consumers you can run on each worker node is limited only by the amount of allowed connections on your Redis server.

Infrastructure

The library leverages ioredis for communication with the Redis server.

Usage

package main

import (
	"github.com/go-redis/redis/v8"
	"context"
	"testing"
	"time"
	"fmt"
	"sync/atomic"
	"github.com/zavitax/redis-ordered-queue-go"
)

func main () {
	var redisOptions = &redis.Options{
		Addr: "127.0.0.1:6379",
		Password: "",
		DB: 0,
	};

	msgCount := int64(0)

	var opt = &redisOrderedQueue.Options{
		RedisOptions: redisOptions,
		BatchSize: 10,
		GroupVisibilityTimeout: time.Second * 60,
		PollingTimeout: time.Second * 10,
		ConsumerCount: 5000,
		RedisKeyPrefix: "{redis-ordered-queue}",
		HandleMessage: func (ctx context.Context, data *interface{}, meta *redisOrderedQueue.MessageMetadata) (error) {
			numMsgs := atomic.AddInt64(&msgCount, 1)

			if (numMsgs % 1000 == 0) {
				fmt.Printf("handleMessage: %v messages processed\n", numMsgs)
			}

			time.Sleep(time.Millisecond * 1500)

			return nil
		},
		HandleInvalidMessage: func (ctx context.Context, data *string) (error) {
			fmt.Printf("handleInvalidMessage: data %v\n", *data)

			return nil
		},
	}

	var client, err = redisOrderedQueue.NewClient(context.TODO(), opt);

	if (err != nil) { t.Errorf("NewClient: %v", err); return }

	metricsLoop := func () {
		opt := &redisOrderedQueue.GetMetricsOptions{
			TopMessageGroupsLimit: 10,
		}

		for {
			met, _ := client.GetMetrics(context.TODO(), opt)
			fmt.Printf("Metrics: g: %v, msgs: %v, conc: %v, topbklg: %v, min: %v, avg: %v, max: %v, %v\n",
				met.TrackedMessageGroups,
				met.VisibleMessages,
				met.WorkingConsumers,
				met.TopMessageGroupsMessageBacklogLength,
				met.MinLatency,
				met.AvgLatency,
				met.MaxLatency,
				met)
			time.Sleep(time.Second * 1)
		}
	};

	go metricsLoop();

	for i := 0; i < 1000; i++ {
		fmt.Printf(".")
		client.Send(context.TODO(), "test message", 1, fmt.Sprintf("GO-GROUP-%v", i));
	}

	client.StartConsumers(context.TODO())
	time.Sleep(time.Second * 60 * 5);
  client.StopConsumers(context.TODO())
  client.Close()
}