Google Pub/Sub #

Google Cloud Pub/Sub is a fully managed asynchronous messaging and data streaming service on Google Cloud Platform. It offers at-least-once delivery guarantees, ordering guarantees (with ordering keys), and native integration with the GCP ecosystem such as Cloud Functions, Dataflow, and BigQuery. Unlike Kafka (self-managed) and RabbitMQ (AMQP protocol), Pub/Sub is a fully managed service — no servers to configure.

Basic Pub/Sub Concepts #

Topic        → a channel for publishing messages
Subscription → a subscription to a topic; each subscription has its own copy of messages
Publisher    → sends messages to a topic
Subscriber   → receives messages from a subscription (pull or push)
Message ID   → the unique ID GCP assigns to each message
Ack ID       → the ID used to acknowledge a message as processed

Visualizing the Google Cloud Pub/Sub Architecture #

Here’s a visualization of publishing messages to a Topic, distributing them to Subscriptions, and the difference between the Pull and Push delivery mechanisms:

flowchart TD
    Pub["Publisher (Go Application)"] -->|"Publish Message"| Topic["GCP Pub/Sub Topic"]
    
    Topic -->|"Message Copy"| SubPull["Subscription A (Pull)"]
    Topic -->|"Message Copy"| SubPush["Subscription B (Push)"]

    SubPull -->|"Pull (Worker Requests)"| Work1["Gopher Worker 1"]
    SubPull -->|"Pull (Worker Requests)"| Work2["Gopher Worker 2"]

    SubPush -->|"Push (GCP Sends)"| Endpoint["HTTPS API Endpoint (Cloud Run / Serverless)"]

Installation #

go get cloud.google.com/go/pubsub

Setting Up the Client #

import "cloud.google.com/go/pubsub"

func newPubSubClient(ctx context.Context, projectID string) (*pubsub.Client, error) {
    // Authentication via Application Default Credentials (ADC):
    // - GOOGLE_APPLICATION_CREDENTIALS env var
    // - gcloud auth application-default login
    // - Service account on GCE/GKE/Cloud Run
    client, err := pubsub.NewClient(ctx, projectID)
    if err != nil {
        return nil, fmt.Errorf("create client: %w", err)
    }
    return client, nil
}

// For a local emulator (development)
func newEmulatorClient(ctx context.Context) (*pubsub.Client, error) {
    // Set env: PUBSUB_EMULATOR_HOST=localhost:8085
    return pubsub.NewClient(ctx, "local-project")
}

Creating Topics and Subscriptions #

func createTopic(ctx context.Context, client *pubsub.Client, topicID string) (*pubsub.Topic, error) {
    // Create the topic if it doesn't exist
    topic, err := client.CreateTopic(ctx, topicID)
    if err != nil {
        // If it already exists, get the existing topic
        if status.Code(err) == codes.AlreadyExists {
            return client.Topic(topicID), nil
        }
        return nil, fmt.Errorf("create topic: %w", err)
    }

    // Topic configuration
    cfg := pubsub.TopicConfig{
        MessageStoragePolicy: pubsub.MessageStoragePolicy{
            AllowedPersistenceRegions: []string{"asia-southeast1"},
        },
        // Schema validation (optional)
        // SchemaSettings: &pubsub.SchemaSettings{...},
    }
    if _, err := topic.Update(ctx, pubsub.TopicConfigToUpdate{
        MessageStoragePolicy: cfg.MessageStoragePolicy,
    }); err != nil {
        log.Printf("Update topic config: %v", err)
    }

    return topic, nil
}

func createSubscription(ctx context.Context, client *pubsub.Client,
    subID, topicID string) (*pubsub.Subscription, error) {

    topic := client.Topic(topicID)

    sub, err := client.CreateSubscription(ctx, subID, pubsub.SubscriptionConfig{
        Topic:              topic,
        AckDeadline:        30 * time.Second,  // processing time limit before redelivery
        RetainAckedMessages: false,

        // Dead letter topic for messages that repeatedly fail
        DeadLetterPolicy: &pubsub.DeadLetterPolicy{
            DeadLetterTopic:     "projects/myproject/topics/dead-letter",
            MaxDeliveryAttempts: 5,
        },

        // Retry policy
        RetryPolicy: &pubsub.RetryPolicy{
            MinimumBackoff: 10 * time.Second,
            MaximumBackoff: 600 * time.Second,
        },

        // Filter — only receive messages with certain attributes
        Filter: `attributes.source = "order-service"`,

        // Ordering — enable if the publisher uses ordering keys
        EnableMessageOrdering: true,
    })
    if err != nil {
        if status.Code(err) == codes.AlreadyExists {
            return client.Subscription(subID), nil
        }
        return nil, fmt.Errorf("create subscription: %w", err)
    }
    return sub, nil
}

Publish — Sending Messages #

func publish(ctx context.Context, topic *pubsub.Topic, data interface{}, attrs map[string]string) error {
    body, err := json.Marshal(data)
    if err != nil {
        return err
    }

    msg := &pubsub.Message{
        Data:       body,
        Attributes: attrs,
        // OrderingKey — messages with the same key are delivered in order
        // (the subscription must have EnableMessageOrdering=true)
        // OrderingKey: "order-" + orderID,
    }

    result := topic.Publish(ctx, msg)

    // Wait for the server confirmation (blocking)
    msgID, err := result.Get(ctx)
    if err != nil {
        return fmt.Errorf("publish: %w", err)
    }

    log.Printf("Message sent, ID: %s", msgID)
    return nil
}

// Publish with an ordering key
func publishOrdered(ctx context.Context, topic *pubsub.Topic,
    orderingKey string, data interface{}) error {

    body, _ := json.Marshal(data)
    result := topic.Publish(ctx, &pubsub.Message{
        Data:        body,
        OrderingKey: orderingKey,
    })

    _, err := result.Get(ctx)
    return err
}

// Batch publish — send many messages, Pub/Sub auto-batches
func publishBatch(ctx context.Context, topic *pubsub.Topic, messages []interface{}) error {
    // Configure the batch settings
    topic.PublishSettings = pubsub.PublishSettings{
        ByteThreshold:  1e6,                    // flush when the batch reaches 1MB
        CountThreshold: 100,                    // flush at 100 messages
        DelayThreshold: 10 * time.Millisecond,  // flush every 10ms
    }

    results := make([]*pubsub.PublishResult, len(messages))
    for i, msg := range messages {
        body, _ := json.Marshal(msg)
        results[i] = topic.Publish(ctx, &pubsub.Message{Data: body})
    }

    // Wait for all publishes to finish
    var errors []error
    for i, r := range results {
        if _, err := r.Get(ctx); err != nil {
            errors = append(errors, fmt.Errorf("message %d: %w", i, err))
        }
    }

    if len(errors) > 0 {
        return fmt.Errorf("%d messages failed to send", len(errors))
    }
    return nil
}

Receive — Receiving Messages (Pull) #

func receive(ctx context.Context, sub *pubsub.Subscription,
    handler func(ctx context.Context, msg *pubsub.Message) error) error {

    // Configure the receive settings
    sub.ReceiveSettings = pubsub.ReceiveSettings{
        MaxExtension:           60 * time.Minute, // max time to extend the ack deadline
        MaxExtensionPeriod:     10 * time.Minute,
        MaxOutstandingMessages: 100,              // max messages processed concurrently
        MaxOutstandingBytes:    1e9,              // 1GB max outstanding data
        NumGoroutines:          5,                // parallel goroutines for receiving
    }

    return sub.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
        log.Printf("Message received: ID=%s, PublishTime=%s",
            msg.ID, msg.PublishTime.Format(time.RFC3339))
        log.Printf("  Attributes: %v", msg.Attributes)
        log.Printf("  OrderingKey: %s", msg.OrderingKey)

        if err := handler(ctx, msg); err != nil {
            log.Printf("Error processing message: %v", err)
            // Nack — Pub/Sub will redeliver per the retry policy
            msg.Nack()
            return
        }

        // Ack — mark the message as successfully processed
        msg.Ack()
    })
}

// Receive with a timeout
func receiveWithTimeout(ctx context.Context, sub *pubsub.Subscription,
    timeout time.Duration, handler func(*pubsub.Message) error) error {

    ctx, cancel := context.WithTimeout(ctx, timeout)
    defer cancel()

    return sub.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
        if err := handler(msg); err != nil {
            msg.Nack()
            return
        }
        msg.Ack()
    })
}

Push Subscriptions — Server Receives via HTTP #

// For push subscriptions, GCP sends an HTTP POST to your endpoint
// Body format: {"message": {"data": "base64...", "attributes": {...}}, "subscription": "..."}

type PushMessage struct {
    Message struct {
        Data        string            `json:"data"`
        Attributes  map[string]string `json:"attributes"`
        MessageID   string            `json:"messageId"`
        PublishTime string            `json:"publishTime"`
    } `json:"message"`
    Subscription string `json:"subscription"`
}

func pubsubPushHandler(w http.ResponseWriter, r *http.Request) {
    var pushMsg PushMessage
    if err := json.NewDecoder(r.Body).Decode(&pushMsg); err != nil {
        http.Error(w, "bad request", http.StatusBadRequest)
        return
    }

    // Decode the base64 data
    data, err := base64.StdEncoding.DecodeString(pushMsg.Message.Data)
    if err != nil {
        http.Error(w, "bad data", http.StatusBadRequest)
        return
    }

    log.Printf("Push message: ID=%s, data=%s",
        pushMsg.Message.MessageID, string(data))

    // Process the message
    if err := processEvent(data, pushMsg.Message.Attributes); err != nil {
        // Return non-200 → Pub/Sub will retry
        http.Error(w, "processing failed", http.StatusInternalServerError)
        return
    }

    // Return 200 → Pub/Sub considers the message acked
    w.WriteHeader(http.StatusNoContent)
}

Complete Example Program — Event Streaming Pipeline #

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"

    "cloud.google.com/go/pubsub"
)

const (
    projectID   = "my-gcp-project"
    topicOrders = "orders"
    subOrders   = "orders-processor"
)

type OrderEvent struct {
    EventID    string    `json:"event_id"`
    Type       string    `json:"type"`
    OrderID    string    `json:"order_id"`
    CustomerID string    `json:"customer_id"`
    Amount     float64   `json:"amount"`
    Items      []string  `json:"items"`
    Timestamp  time.Time `json:"timestamp"`
}

// EventPublisher sends events to Pub/Sub
type EventPublisher struct {
    topic *pubsub.Topic
}

func NewEventPublisher(ctx context.Context, client *pubsub.Client, topicID string) (*EventPublisher, error) {
    topic := client.Topic(topicID)
    ok, err := topic.Exists(ctx)
    if err != nil {
        return nil, err
    }
    if !ok {
        topic, err = client.CreateTopic(ctx, topicID)
        if err != nil {
            return nil, err
        }
    }

    topic.PublishSettings = pubsub.PublishSettings{
        CountThreshold: 100,
        DelayThreshold: 10 * time.Millisecond,
    }

    return &EventPublisher{topic: topic}, nil
}

func (p *EventPublisher) Publish(ctx context.Context, event OrderEvent) error {
    data, err := json.Marshal(event)
    if err != nil {
        return err
    }

    result := p.topic.Publish(ctx, &pubsub.Message{
        Data: data,
        Attributes: map[string]string{
            "event_type": event.Type,
            "source":     "order-service",
            "version":    "1.0",
        },
        // Use the OrderID as the ordering key — events for the same order are always in order
        OrderingKey: event.OrderID,
    })

    msgID, err := result.Get(ctx)
    if err != nil {
        return fmt.Errorf("publish: %w", err)
    }
    log.Printf("[PUB] event=%s order=%s msgID=%s", event.Type, event.OrderID, msgID)
    return nil
}

func (p *EventPublisher) Close() {
    p.topic.Stop()
}

// EventProcessor processes events from Pub/Sub
type EventProcessor struct {
    sub     *pubsub.Subscription
    stats   map[string]int
    statsMu sync.Mutex
}

func NewEventProcessor(ctx context.Context, client *pubsub.Client, subID, topicID string) (*EventProcessor, error) {
    topic := client.Topic(topicID)

    sub := client.Subscription(subID)
    ok, err := sub.Exists(ctx)
    if err != nil {
        return nil, err
    }
    if !ok {
        sub, err = client.CreateSubscription(ctx, subID, pubsub.SubscriptionConfig{
            Topic:                 topic,
            AckDeadline:           30 * time.Second,
            EnableMessageOrdering: true,
        })
        if err != nil {
            return nil, err
        }
    }

    sub.ReceiveSettings = pubsub.ReceiveSettings{
        MaxOutstandingMessages: 50,
        NumGoroutines:          3,
    }

    return &EventProcessor{sub: sub, stats: make(map[string]int)}, nil
}

func (p *EventProcessor) Start(ctx context.Context) error {
    log.Println("[SUB] Started processing events...")

    return p.sub.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
        var event OrderEvent
        if err := json.Unmarshal(msg.Data, &event); err != nil {
            log.Printf("[SUB] ERROR unmarshal: %v", err)
            msg.Nack()
            return
        }

        log.Printf("[SUB] ID=%s type=%s order=%s amount=Rp%.0f",
            msg.ID[:8]+"...", event.Type, event.OrderID, event.Amount)

        if err := p.process(event); err != nil {
            log.Printf("[SUB] FAILED to process %s: %v", event.OrderID, err)
            msg.Nack()
            return
        }

        p.statsMu.Lock()
        p.stats[event.Type]++
        p.statsMu.Unlock()

        msg.Ack()
    })
}

func (p *EventProcessor) process(event OrderEvent) error {
    switch event.Type {
    case "order.created":
        fmt.Printf("  → Create order record %s (Rp%.0f)\n", event.OrderID, event.Amount)
        time.Sleep(50 * time.Millisecond)
    case "order.paid":
        fmt.Printf("  → Mark paid and send a notification: %s\n", event.OrderID)
        time.Sleep(30 * time.Millisecond)
    case "order.shipped":
        fmt.Printf("  → Update tracking: %s\n", event.OrderID)
        time.Sleep(20 * time.Millisecond)
    case "order.completed":
        fmt.Printf("  → Complete the order and update stats: %s\n", event.OrderID)
        time.Sleep(40 * time.Millisecond)
    }
    return nil
}

func (p *EventProcessor) PrintStats() {
    p.statsMu.Lock()
    defer p.statsMu.Unlock()
    fmt.Println("\n=== Processing Statistics ===")
    total := 0
    for t, c := range p.stats {
        fmt.Printf("  %-20s: %d events\n", t, c)
        total += c
    }
    fmt.Printf("  %-20s: %d events\n", "TOTAL", total)
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())

    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    go func() {
        <-sigCh
        log.Println("Received signal, shutting down...")
        cancel()
    }()

    // Set up the client
    client, err := pubsub.NewClient(ctx, projectID)
    if err != nil {
        log.Fatal("Client:", err)
    }
    defer client.Close()

    // Publisher
    publisher, err := NewEventPublisher(ctx, client, topicOrders)
    if err != nil {
        log.Fatal("Publisher:", err)
    }
    defer publisher.Close()

    // Processor
    processor, err := NewEventProcessor(ctx, client, subOrders, topicOrders)
    if err != nil {
        log.Fatal("Processor:", err)
    }

    // Start the processor in a goroutine
    go func() {
        if err := processor.Start(ctx); err != nil && ctx.Err() == nil {
            log.Printf("Processor error: %v", err)
        }
    }()

    // Publish events
    fmt.Println("=== Sending Order Events ===")
    orders := []struct {
        id     string
        events []string
        amount float64
        items  []string
    }{
        {"ORD-001", []string{"order.created", "order.paid", "order.shipped", "order.completed"},
            1_500_000, []string{"Laptop", "Mouse"}},
        {"ORD-002", []string{"order.created", "order.paid"},
            350_000, []string{"Keyboard"}},
        {"ORD-003", []string{"order.created"},
            85_000, []string{"T-Shirt"}},
    }

    for _, order := range orders {
        for i, evtType := range order.events {
            event := OrderEvent{
                EventID:    fmt.Sprintf("evt-%s-%d", order.id, i),
                Type:       evtType,
                OrderID:    order.id,
                CustomerID: "CUST-42",
                Amount:     order.amount,
                Items:      order.items,
                Timestamp:  time.Now(),
            }
            if err := publisher.Publish(ctx, event); err != nil {
                log.Printf("Failed to publish: %v", err)
            }
            time.Sleep(100 * time.Millisecond)
        }
    }

    // Wait for processing
    fmt.Println("\n=== Processing Events ===")
    select {
    case <-ctx.Done():
    case <-time.After(5 * time.Second):
        cancel()
    }

    processor.PrintStats()
    fmt.Println("Done.")
}

Summary #

Pub/Sub is a fully managed service on GCP — no brokers, replication, or partitions to manage.
Topics for publishing; Subscriptions for receiving — each subscription gets its own copy of messages.
topic.Publish() returns a PublishResult — call .Get(ctx) for the delivery confirmation.
PublishSettings controls automatic batching — adjust CountThreshold and DelayThreshold.
OrderingKey guarantees message order for the same key — enable EnableMessageOrdering on the subscription.
msg.Ack() confirms successful processing; msg.Nack() triggers redelivery per the retry policy.
ReceiveSettings.NumGoroutines for parallelism; MaxOutstandingMessages for backpressure.
Dead letter topics for messages that fail after MaxDeliveryAttempts times.
Filters on subscriptions (attributes.source = "order-service") for filtering without extra code.
Push subscriptions for serverless (Cloud Functions, Cloud Run) — GCP POSTs to your HTTP endpoint.

← Previous: Amazon SQS Next: Redis →