Queues and Messaging
Send and receive SQS messages, keep the visibility timeout longer than the work, and make the consumer idempotent.
out, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{QueueUrl: aws.String(q), MaxNumberOfMessages: 10, WaitTimeSeconds: 20})A queue holds messages between two programs. One sends, another receives later. The sender does not wait for the work to finish, and a slow consumer cannot slow the sender down. SQS is the AWS queue service, and .../service/sqs is its module.
Send, receive, delete
SendMessage puts one message on a queue. The body is a string, usually JSON. ReceiveMessage asks for up to 10 messages at once. Set WaitTimeSeconds to 20 and the call waits up to 20 seconds for something to arrive instead of returning empty immediately. That is long polling, and it costs far fewer requests than asking again in a tight loop.
Receiving does not remove a message. It hides it for the length of the visibility timeout so no other consumer sees it. You call DeleteMessage with the message's receipt handle once the work is done. If your process dies first, the message becomes visible again and another consumer picks it up. That recovery is the point of the design.
Idempotent consumers
A standard SQS queue delivers each message at least once, which means occasionally twice. So a consumer must be idempotent: handling a message twice must leave the same result as handling it once. Choose a key, such as an order id inside the body, record it when you do the work, and skip any message whose key is already recorded. Record it in the same database transaction as the work, or the two can disagree after a crash.
Read the body with aws.ToString(m.Body), the shorthand for *m.Body that returns "" instead of panicking when the pointer is nil.
Example
package main
import (
"context"
"fmt"
"log"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
)
// deleter names the one method of *sqs.Client that consume needs, which is
// also what lets the fake below stand in for the real client.
type deleter interface {
DeleteMessage(context.Context, *sqs.DeleteMessageInput, ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error)
}
type fakeSQS struct{}
func (fakeSQS) DeleteMessage(context.Context, *sqs.DeleteMessageInput, ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error) {
return &sqs.DeleteMessageOutput{}, nil
}
// consume charges each order once, then deletes the message so that SQS
// stops delivering it.
func consume(ctx context.Context, c deleter, queue string, msgs []types.Message, done map[string]bool) error {
for _, m := range msgs {
id := aws.ToString(m.Body) // the order id doubles as the dedupe key
if done[id] {
fmt.Println("already charged:", id)
} else {
fmt.Println("charging:", id)
done[id] = true
}
// Delete only once the work is safely finished.
_, err := c.DeleteMessage(ctx, &sqs.DeleteMessageInput{
QueueUrl: aws.String(queue),
ReceiptHandle: m.ReceiptHandle,
})
if err != nil {
return fmt.Errorf("delete %s: %w", id, err)
}
}
return nil
}
func main() {
// The same order arrives twice: a redelivery after a slow first attempt.
msgs := []types.Message{
{Body: aws.String("order-7"), ReceiptHandle: aws.String("rh-1")},
{Body: aws.String("order-7"), ReceiptHandle: aws.String("rh-2")},
}
err := consume(context.Background(), fakeSQS{}, "https://queue.example/orders", msgs, map[string]bool{})
if err != nil {
log.Fatal(err)
}
}
// Output:
// charging: order-7
// already charged: order-7When to use it
- Move slow work, such as sending an email or rendering a PDF, out of the request and into a worker.
- Absorb a burst of signups so a rate-limited third-party API is never called faster than it allows.
- Keep retrying a failed payment capture by leaving the message on the queue until it succeeds.
- Route messages that keep failing into a dead-letter queue so a human can look at them.
More examples
Sending a message as JSON
This is the producer half, and it is deliberately dull. SQS only carries text, so the event becomes JSON first. Define the message as a struct rather than building the JSON by hand: the consumer can then decode into the same struct, and a renamed field is a compile error in both halves instead of a silent nil. Keep the body small, under the 1 MiB limit, and put a large payload in S3 with the key in the message.
type orderPaid struct {
OrderID string `json:"order_id"`
Pence int `json:"pence"`
}
func publish(ctx context.Context, c *sqs.Client, queueURL string, ev orderPaid) error {
body, err := json.Marshal(ev)
if err != nil {
return err
}
_, err = c.SendMessage(ctx, &sqs.SendMessageInput{
QueueUrl: aws.String(queueURL),
MessageBody: aws.String(string(body)),
})
if err != nil {
return fmt.Errorf("send %s: %w", ev.OrderID, err)
}
return nil
}The receive loop
The three numbers are the whole policy. MaxNumberOfMessages is 10, the most SQS allows per call, so one request fetches a batch. WaitTimeSeconds is 20, also the maximum, so an idle worker makes three requests a minute instead of thousands. VisibilityTimeout is 60, so this batch is hidden for a minute: it must comfortably exceed how long handle takes, or the message is handed to a second worker while the first is still on it. Note what happens on a handler error: nothing is deleted, so the message returns by itself.
// run polls until the context is cancelled, which is how shutdown reaches it.
func run(ctx context.Context, c *sqs.Client, queueURL string) error {
for {
out, err := c.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
QueueUrl: aws.String(queueURL),
MaxNumberOfMessages: 10,
WaitTimeSeconds: 20,
VisibilityTimeout: 60,
})
if err != nil {
if ctx.Err() != nil {
return ctx.Err() // cancelled mid-wait: a clean stop, not a fault
}
return fmt.Errorf("receive: %w", err)
}
for _, m := range out.Messages {
if err := handle(ctx, c, queueURL, m); err != nil {
// Leave it on the queue: it reappears after 60 seconds.
log.Printf("message %s: %v", aws.ToString(m.MessageId), err)
}
}
}
}Deduplicating in the database
Use this instead of an in-memory map as soon as more than one copy of the worker runs, which in a cluster is always. The database is the only place the two copies can agree. ON CONFLICT DO NOTHING turns the duplicate into zero affected rows rather than an error, so the caller gets a plain yes or no. Because the insert and the work share one transaction, a crash halfway cannot leave the key recorded for work that never happened.
// claim inserts the key and reports whether this attempt won the race.
// Called inside the same transaction as the work it guards, so the two
// either both commit or both roll back.
func claim(ctx context.Context, tx *sql.Tx, key string) (bool, error) {
// Table: processed(key TEXT PRIMARY KEY). The primary key rejects a
// second insert of the same value.
res, err := tx.ExecContext(ctx,
`INSERT INTO processed (key) VALUES ($1) ON CONFLICT (key) DO NOTHING`, key)
if err != nil {
return false, err
}
n, err := res.RowsAffected()
if err != nil {
return false, err
}
return n == 1, nil // 0 rows means someone already processed this key
}
Discussion