Getting Started #

In this article you are going to learn how to build queues using Redis and Go.

Project Setup #

Create a new Go Project.

mkdir go-redis-queues
cd go-redis-queues
go mod init github.com/gurleensethi/go-redis-queues

Install the go-redis package.

go get github.com/redis/go-redis/v9

Run redis locally (I am going to use docker to run a redis container).

docker run --rm --name redis -p 6379:6379 -d redis

Connecting to Redis #

package main

import (
	"context"

	"github.com/redis/go-redis/v9"
)

func main() {
    ctx := context.Background()

    client := redis.NewClient(&redis.Options{
        Addr:     "127.0.0.1:6379",
        Password: "",
        DB:       0,
    })

    _, err := client.Ping(ctx).Result()
    if err != nil {
        panic(err)
    }
}
main.go

Producer #

You are going to write a function that connects to Redis and pushes data into the queue every second.

Redis List is going to serve as the underlying data structure to implement queues.

You are going to use LPush (add the item to the left side of the array) to "enqueue".

client := redis.NewClient(&redis.Options{
	Addr:     "127.0.0.1:6379",
	Password: "",
	DB:       0,
})

client.LPush(ctx, "queue", 100).Result()

To push an item every second, you are going to use time.Ticker.

ticker := time.NewTicker(time.Second * 1)

for {
   select {
      case <- ticker.C:
      // Push data to queue
   }
}

Putting it all together you get the following.

package main

import (
	"context"
	"fmt"
	"math/rand"
	"time"

	"github.com/redis/go-redis/v9"
)

func main() {
	ctx := context.Background()

	go producer(ctx)

    // Prevent the program to exit
	time.Sleep(time.Second * 100)
}


// producer pushes a new random item to the queue every second
func producer(ctx context.Context) {
	client := redis.NewClient(&redis.Options{
		Addr:     "127.0.0.1:6379",
		Password: "",
		DB:       0,
	})

	ticker := time.NewTicker(time.Second * 1)

	for {
		select {
		case <-ctx.Done():
			ticker.Stop()
			return
		case <-ticker.C:
			// Push a random integer in the queue
			v := rand.Int()

			_, err := client.LPush(ctx, "queue", v).Result()
			if err != nil {
				fmt.Printf("error pushing to queue: %w", err)
			}
		}
	}
}
main.go

Consumer #

Similar to the producer function, you are going to write consumer function that reads from the queue and prints it out.

For brevity you are going to write the consumer in the same file, in a real world scenario this would ideally be a separate program.

You are going to use BRPop to pop and item from the left side of the Redis List. BRPop is the blocking variant of RPop, basically if there is no item in the queue the command will wait for items to appear.

client := redis.NewClient(&redis.Options{
	Addr:     "127.0.0.1:6379",
	Password: "",
	DB:       0,
})

client.BRPop(ctx, 0, "queue").Result()

You are going to pop and item inside a for loop.

for {
    result, err := client.BRPop(ctx, 0, "queue").Result()
}

Putting it all together you get the following.

package main

import (
	"context"
	"fmt"
	"math/rand"
	"time"

	"github.com/redis/go-redis/v9"
)

func main() {
	ctx := context.Background()

	go producer(ctx)

	go consumer(ctx, "1")
	go consumer(ctx, "2")
	go consumer(ctx, "3")

	time.Sleep(time.Second * 100)
}

func producer() {
    ...
}


func consumer(ctx context.Context, consumerName string) {
	client := redis.NewClient(&redis.Options{
		Addr:     "127.0.0.1:6379",
		Password: "",
		DB:       0,
	})

	for {
		result, err := client.BRPop(ctx, 0, "queue").Result()
		if err != nil {
			fmt.Printf("[%s] error popping from queue: %w\n", consumerName, err)
			continue
		}

		fmt.Printf("[%s] received: %v\n", consumerName, result)

		// Wait a random amount of time before popping the next item
		time.Sleep(time.Duration(rand.Intn(5)) * time.Second)
	}
}
main.go

You find the complete code in this github repository.