PGQ - go queues on top of postgres

PGQ is a Go package that provides a queuing mechanism for your Go applications. It is built on top of the postgres database and enables developers to implement efficient and reliable, but simple message queues for their services composed architecture using the familiar postgres infrastructure.

Features

• Postgres-backed: Leverages the power of SQL to store and manage queues.
• Reliable: Guarantees message persistence and delivery, even if facing the various failures.
• Transactional: Supports transactional message handling, ensuring consistency.
• Simple usage: Provides a clean and easy-to-use API for interacting with the queue.
• Efficient: Optimized for medium to long jobs durations.
• Scheduled Messages: Schedule messages to be processed at a specific future time.

Why PGQ?

• Postgres: The postgres just works and is feature rich, scalable and performant.
• SQL: You are already familiar with SQLm right? No need to learn anything new.
• Stack: You do not have to maintain/manage/administer/patch any additional message broker component.
• Simplicity: The pgq is a simple and straightforward solution for your messaging needs.
• Usability: The pgq can be used for many scenarios.

Of course, you can implement your own implementation or use other messaging technologies, but why would you do that when you can use the postgres which is already there and is a proven technology?

See the benefits using pgq in the following video:

When to pick PGQ?

Even though there are other great technologies and tools for complex messaging including the robust routing configuration, sometimes you do not need it, and you can be just fine with the simpler tooling.

Pick pgq if you: - need to distribute the traffic fairly among your app replicas - need to protect your services from overload - want the out-of-a-box observability of your queues - want to use SQL for managing your queues - already use postgres and you want to keep your tech stack simple - don't want to manage another technology and learn its specifics - need to schedule messages to be processed at a specific time

No need to bring the new technology to your existing stack when you can be pretty satisfied with postgres. Write the consumers and publishers in various languages with the simple idea behind - use postgres table as a queue. While using pgq you have a superb observability of the queue.
You can easily see the payloads of the messages waiting to be processed, but moreover payloads of both the currently being processed and already processed messages. You can get the processing results, duration and other statistics pretty simply. As the pgq queue table contains the records of already processed jobs too, you already have out of the box the historical statistics of all messages and can view it effortlessly by using simple SQL queries.

Pgq is intended to replace the specialized message brokers in environments where you already use postgres, and you want clean, simple and straightforward communication among your services.

Basic principles

• Every queue is the single postgres table.
• You maintain the table on your own. You can extend it as you need.
• Publishers add new rows to the queue table.
• Consumers update the pgq mandatory fields of the rows in the queue table.

Installation

To install PGQ, use the go get command:

go get go.dataddo.com/pgq@latest

Setup

Prerequisites:

In order to make the pgq functional, there must exist the postgres table with all the necessary pgq fields. You can create the table on your own with classic CREATE TABLE ..., or you can use the query generator to generate the query for you. The generated query creates the queue table alongside with indexes which improve the consumer queries performance.

You usually run the setup commands just once during the queue setup.

package main

import (
    "fmt"
    "go.dataddo.com/pgq/x/schema"
)

func main() {
    queueName := "my_queue"

    // create string contains the "CREATE TABLE queueName ..." 
    // which you may use for table and indexes creation.
    create := schema.GenerateCreateTableQuery(queueName)
    fmt.Println(create)

    // You may also use the "GenerateDropTableQuery" for dropping all the pgq artifacts (down migration)
}

Usage

Publishing the message

package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "fmt"

    _ "github.com/jackc/pgx/v5/stdlib"

    "go.dataddo.com/pgq"
)

func main() {
    postgresDSN := "your_postgres_dsn"
    queueName := "your_queue_name"

    // create a new postgres connection 
    db, err := sql.Open("pgx", postgresDSN)
    if err != nil {
        panic(err.Error())
    }
    defer db.Close()

    // create the publisher which may be reused for multiple messages
    // you may pass the optional PublisherOptions when creating it
    publisher := pgq.NewPublisher(db)

    // publish the message to the queue
    // provide the payload which is the JSON object
    // and optional metadata which is the map[string]string
    msg := &pgq.MessageOutgoing{
        Payload: json.RawMessage(`{"foo":"bar"}`),
    }
    msgId, err := publisher.Publish(context.Background(), queueName, msg)
    if err != nil {
        panic(err.Error())
    }

    fmt.Println("Message published with ID:", msgId)
}

After the message is successfully published, you can see the new row with given msgId in the queue table.

Publisher options

Very often you want some metadata to be part of the message, so you can filter the messages in the queue table by it. Metadata can be any additional information you think is worth to be part of the message, but you do not want to be part of the payload. It can be the publisher app name/version, payload schema version, customer identifiers etc etc.

You can simply attach the metadata to single message by:

metadata := pgq.Metadata{
    "publisherHost": "localhost",
    "payloadVersion": "v1.0"
}

or you can configure the publisher to attach the metadata to all messages it publishes:

opts := []pgq.PublisherOption{
    pgq.WithMetaInjectors(
        pgq.StaticMetaInjector(
            pgq.Metadata{
                "publisherHost": "localhost",
                "publisherVersion": "commitRSA"
            }
        ),
    ),
},

publisher := pgq.NewPublisher(db, opts)
metadata := pgq.Metadata{
    "payloadVersion": "v1.0" // message specific meta field
}

Consuming the messages

package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "fmt"
    "go.dataddo.com/pgq"
    _ "github.com/jackc/pgx/v5/stdlib"
)

func main() {
    postgresDSN := "your_postgres_dsn"
    queueName := "your_queue_name"

    // create a new postgres connection and publisher 
    db, err := sql.Open("pgx", postgresDSN)
    if err != nil {
        panic(err.Error())
    }
    defer db.Close()

    // create the consumer which gets attached to handling function we defined above
    h := &handler{}
    consumer, err := pgq.NewConsumer(db, queueName, h)
    if err != nil {
        panic(err.Error())
    }

    err = consumer.Run(context.Background())
    if err != nil {
        panic(err.Error())
    }
}

// we must specify the message handler, which implements simple interface
type handler struct {}
func (h *handler) HandleMessage(_ context.Context, msg *pgq.MessageIncoming) (processed bool, err error) {
    fmt.Println("Message payload:", string(msg.Payload))
    return true, nil
}

Consumer options

You can configure the consumer by passing the optional ConsumeOptions when creating it.

consumer, err := NewConsumer(db, queueName, handler,
        WithLogger(slog.New(slog.NewTextHandler(&tbWriter{tb: t}, &slog.HandlerOptions{Level: slog.LevelDebug}))),
        WithLockDuration(10 * time.Minute),
        WithPollingInterval(2 * time.Second),
        WithMaxParallelMessages(1),
        WithMetrics(noop.Meter{}),
    )

For more detailed usage examples and API documentation, please refer to the Dataddo pgq GoDoc page.

Message

The message is the essential structure for communication between services using pgq. The message struct matches the postgres table schema. You can modify the table structure on your own by adding extra columns, but pgq depends on following mandatory fields only: