Google Pub/Sub #

Google Cloud Pub/Sub is a managed messaging service from Google Cloud Platform designed for global scale with low latency. Similar to Amazon SQS, there’s no infrastructure to manage — but Pub/Sub has a model closer to Kafka in one important aspect: one topic can have many subscriptions, each receiving an independent copy of all messages, rather than sharing messages like SQS. In Rust, the community google-cloud-pubsub crate provides ergonomic async access. This article covers authentication, topic and subscription management, publishing, consuming with two modes (pull and streaming), and common architectural patterns in the GCP ecosystem.

Pub/Sub Architecture #

flowchart LR
    P1["Publisher A"] --> T["Topic\npesanan-events"]
    P2["Publisher B"] --> T

    T --> S1["Subscription 1\npemrosesan-pesanan\n(pull mode)"]
    T --> S2["Subscription 2\nnotifikasi-email\n(push mode → Cloud Run)"]
    T --> S3["Subscription 3\nanalytics-stream\n(BigQuery subscription)"]

    S1 --> C1["Consumer\n(GKE pod)"]
    S2 --> C2["Cloud Run\nService"]
    S3 --> C3["BigQuery\nTable"]

Concept	Explanation
Topic	Message channel — publishers send here
Subscription	How consumers read from a topic
Pull	The consumer actively requests messages from Pub/Sub
Push	Pub/Sub pushes messages to an HTTP endpoint
Ack	Confirmation that a message was successfully processed
Ack Deadline	Time limit to ack before the message reappears

Key difference from SQS: every subscription receives a copy of all messages on the topic. If there are 3 subscriptions, each message is delivered 3 times to different consumers. In SQS, one message is consumed by only one consumer.

Installation #

[dependencies]
google-cloud-pubsub = "0.22"
google-cloud-gax = "0.18"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
base64 = "0.21"  # Pub/Sub messages are base64-encoded

Authentication #

Google Cloud uses Application Default Credentials (ADC). The most common ways:

# Local development — log in with a Google account
gcloud auth application-default login

# On GCE/GKE/Cloud Run — automatically uses Workload Identity or a service account
# No extra configuration needed

# Or use a service account key file
export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json

use google_cloud_pubsub::client::{Client, ClientConfig};

async fn buat_client(project_id: &str) -> Result<Client, Box<dyn std::error::Error>> {
    let config = ClientConfig::default()
        .with_project_id(project_id);

    // ADC is automatically read from the environment
    let client = Client::new(config).await?;
    Ok(client)
}

// For local development with the Pub/Sub emulator
async fn buat_client_emulator(project_id: &str) -> Result<Client, Box<dyn std::error::Error>> {
    // Set the environment variable before running:
    // export PUBSUB_EMULATOR_HOST=localhost:8085
    std::env::set_var("PUBSUB_EMULATOR_HOST", "localhost:8085");

    let config = ClientConfig::default()
        .with_project_id(project_id);

    let client = Client::new(config).await?;
    Ok(client)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let project_id = std::env::var("GCP_PROJECT_ID")
        .unwrap_or_else(|_| "my-project".to_string());

    let client = buat_client(&project_id).await?;
    println!("Connected to Google Cloud Pub/Sub (project: {})", project_id);

    Ok(())
}

Topic and Subscription Management #

use google_cloud_pubsub::client::Client;
use google_cloud_pubsub::topic::TopicConfig;
use google_cloud_pubsub::subscription::{SubscriptionConfig, RetryPolicy};
use std::time::Duration;

async fn setup_topic_dan_subscription(
    client: &Client,
    nama_topic: &str,
    nama_subscription: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    // Create the topic
    let topic = client.create_topic(
        nama_topic,
        None,  // default TopicConfig
        None,  // retry config
    ).await?;
    println!("Topic created: {}", topic.fully_qualified_name());

    // Create a subscription with configuration
    let sub_config = SubscriptionConfig {
        // Ack deadline: how many seconds to ack before the message reappears
        ack_deadline_seconds: 60,
        // Retain acknowledged messages for how long
        retain_acked_messages: false,
        // Message retention duration
        message_retention_duration: Some(Duration::from_secs(86400)), // 1 day
        // Retry policy
        retry_policy: Some(RetryPolicy {
            minimum_backoff: Duration::from_secs(10),
            maximum_backoff: Duration::from_secs(600),
        }),
        ..Default::default()
    };

    let subscription = client.create_subscription(
        nama_subscription,
        nama_topic,
        sub_config,
        None,
    ).await?;
    println!("Subscription created: {}", subscription.fully_qualified_name());

    Ok(())
}

// Create a subscription with a Dead Letter Topic
async fn setup_dengan_dead_letter(
    client: &Client,
    nama_topic: &str,
    nama_sub: &str,
    nama_dlq_topic: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    use google_cloud_pubsub::subscription::DeadLetterPolicy;

    // Create the DLQ topic first
    client.create_topic(nama_dlq_topic, None, None).await?;

    // Create the subscription with a dead letter policy
    let project_id = std::env::var("GCP_PROJECT_ID").unwrap_or_default();
    let dlq_topic_name = format!(
        "projects/{}/topics/{}",
        project_id, nama_dlq_topic
    );

    let sub_config = SubscriptionConfig {
        ack_deadline_seconds: 30,
        dead_letter_policy: Some(DeadLetterPolicy {
            dead_letter_topic: dlq_topic_name,
            max_delivery_attempts: 5, // move to the DLQ after 5 failures
        }),
        ..Default::default()
    };

    client.create_subscription(nama_sub, nama_topic, sub_config, None).await?;
    println!("Subscription with DLQ created: {}", nama_sub);

    Ok(())
}

Publishing Messages #

use google_cloud_pubsub::client::Client;
use google_cloud_pubsub::publisher::PublisherConfig;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

#[derive(Debug, Serialize, Deserialize, Clone)]
struct EventPesanan {
    pub id: u64,
    pub pengguna_id: u64,
    pub total: f64,
    pub status: String,
    pub dibuat_pada: String,
}

async fn publish_pesan(
    client: &Client,
    nama_topic: &str,
    event: &EventPesanan,
) -> Result<String, Box<dyn std::error::Error>> {
    let topic = client.topic(nama_topic);

    // Publisher configuration
    let publisher = topic.new_publisher(Some(PublisherConfig {
        // Batching: wait for up to 100 messages or 10ms before sending
        flush_interval: std::time::Duration::from_millis(10),
        bundle_size: 100,
        ..Default::default()
    }));

    // Encode the payload to bytes
    let payload = serde_json::to_vec(event)?;

    // Message attributes — metadata that can be filtered at the subscription
    let mut attributes = HashMap::new();
    attributes.insert("event_type".to_string(), event.status.clone());
    attributes.insert("source".to_string(), "api-server".to_string());
    attributes.insert("version".to_string(), "v1".to_string());

    // Ordering key: messages with the same key are guaranteed ordered at the subscription
    let ordering_key = format!("pengguna-{}", event.pengguna_id);

    let pesan = google_cloud_pubsub::publisher::PubsubMessage {
        data: payload,
        attributes,
        ordering_key,  // leave empty if ordering isn't needed
        ..Default::default()
    };

    // Send — returns a Future that resolves to the message ID
    let awaiter = publisher.publish(pesan).await;

    // Wait for server confirmation
    let msg_id = awaiter.get().await?;
    println!("Message sent with ID: {}", msg_id);

    // Flush all buffered messages before shutdown
    publisher.shutdown().await;

    Ok(msg_id)
}

// Publish many messages in parallel
async fn publish_batch(
    client: &Client,
    nama_topic: &str,
    events: Vec<EventPesanan>,
) -> Result<Vec<String>, Box<dyn std::error::Error>> {
    let topic = client.topic(nama_topic);
    let publisher = topic.new_publisher(None);

    // Send all messages without waiting (concurrent)
    let awaiters: Vec<_> = events.iter().map(|event| {
        let payload = serde_json::to_vec(event).unwrap();
        let pesan = google_cloud_pubsub::publisher::PubsubMessage {
            data: payload,
            ..Default::default()
        };
        publisher.publish(pesan)
    }).collect();

    // Collect the publisher futures
    let awaiters_resolved: Vec<_> = futures::future::join_all(awaiters).await;

    // Wait for all confirmations
    let mut msg_ids = Vec::new();
    for awaiter in awaiters_resolved {
        let id = awaiter.get().await?;
        msg_ids.push(id);
    }

    publisher.shutdown().await;
    println!("Batch publish: {} messages succeeded", msg_ids.len());

    Ok(msg_ids)
}

Pull — Manually Pulling Messages #

use google_cloud_pubsub::client::Client;
use google_cloud_pubsub::subscription::ReceiveConfig;

async fn pull_pesan(
    client: &Client,
    nama_subscription: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let subscription = client.subscription(nama_subscription);

    // Pull messages with a maximum limit
    let pesan_list = subscription
        .pull(10, None)  // max 10 messages, no timeout
        .await?;

    for pesan in pesan_list {
        let data = std::str::from_utf8(&pesan.message.data).unwrap_or("");
        let msg_id = &pesan.message.message_id;

        println!("Message ID: {}", msg_id);

        // Read attributes
        for (k, v) in &pesan.message.attributes {
            println!("  {}: {}", k, v);
        }

        // Parse the payload
        match serde_json::from_str::<EventPesanan>(data) {
            Ok(event) => {
                match proses_event(&event).await {
                    Ok(_) => {
                        // Acknowledge — the message is removed from the subscription
                        pesan.ack().await?;
                        println!("Event #{} processed and acked", event.id);
                    }
                    Err(e) => {
                        eprintln!("Failed to process event #{}: {}", event.id, e);
                        // Nack — the message reappears after the ack deadline
                        pesan.nack().await?;
                    }
                }
            }
            Err(e) => {
                eprintln!("Failed to parse message: {}", e);
                pesan.nack().await?;
            }
        }
    }

    Ok(())
}

async fn proses_event(event: &EventPesanan) -> Result<(), String> {
    println!("Processing order #{} - {}", event.id, event.status);
    Ok(())
}

Streaming Pull — Continuous Consumption #

The most common mode for always-running consumers:

use google_cloud_pubsub::client::Client;
use google_cloud_pubsub::subscription::ReceiveConfig;
use futures::StreamExt;

async fn streaming_consumer(
    client: &Client,
    nama_subscription: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let subscription = client.subscription(nama_subscription);

    println!("Streaming consumer active: {}", nama_subscription);

    let config = ReceiveConfig {
        // Number of worker goroutines to process messages in parallel
        worker_count: 4,
        ..Default::default()
    };

    // receive: opens a continuous streaming pull
    subscription.receive(
        move |pesan, ack_handler| {
            async move {
                let data = std::str::from_utf8(&pesan.data).unwrap_or("{}");
                let msg_id = &pesan.message_id;

                println!("Received [{}]: {}...", msg_id, &data[..data.len().min(80)]);

                match serde_json::from_str::<EventPesanan>(data) {
                    Ok(event) => {
                        match proses_event(&event).await {
                            Ok(_) => {
                                ack_handler.ack().await;
                            }
                            Err(e) => {
                                eprintln!("Error: {}", e);
                                ack_handler.nack().await;
                            }
                        }
                    }
                    Err(_) => {
                        // Message can't be parsed — ack to avoid an infinite loop
                        eprintln!("Invalid message, acking to skip");
                        ack_handler.ack().await;
                    }
                }
            }
        },
        None,    // CancellationToken — for graceful shutdown
        Some(config),
    )
    .await?;

    Ok(())
}

Extending the Ack Deadline #

For messages that need longer processing time:

use google_cloud_pubsub::client::Client;
use google_cloud_pubsub::subscription::ReceiveConfig;

async fn consumer_dengan_keepalive(
    client: &Client,
    nama_subscription: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let subscription = client.subscription(nama_subscription);

    subscription.receive(
        move |pesan, ack_handler| {
            async move {
                let data = std::str::from_utf8(&pesan.data).unwrap_or("{}");

                // Spawn a background task to extend the ack deadline
                let handler_clone = ack_handler.clone();
                let keepalive = tokio::spawn(async move {
                    loop {
                        tokio::time::sleep(std::time::Duration::from_secs(30)).await;
                        // Extend the ack deadline by 60 seconds
                        handler_clone.modify_ack_deadline(60).await;
                        println!("Ack deadline extended");
                    }
                });

                // Simulate a long processing time
                tokio::time::sleep(std::time::Duration::from_secs(90)).await;

                keepalive.abort();
                ack_handler.ack().await;
                println!("Long message processed");
            }
        },
        None,
        None,
    )
    .await?;

    Ok(())
}

Ordering Keys — Order Per Entity #

Messages with the same ordering key are guaranteed ordered within one subscription:

async fn publish_dengan_ordering(
    client: &Client,
    nama_topic: &str,
    pengguna_id: u64,
    events: Vec<EventPesanan>,
) -> Result<(), Box<dyn std::error::Error>> {
    let topic = client.topic(nama_topic);

    // IMPORTANT: the subscription must be created with enable_message_ordering = true
    let publisher = topic.new_publisher(None);

    for event in &events {
        let payload = serde_json::to_vec(event)?;
        let pesan = google_cloud_pubsub::publisher::PubsubMessage {
            data: payload,
            // All events for the same user use the same key
            // → order guaranteed within a subscription that enables ordering
            ordering_key: format!("user-{}", pengguna_id),
            ..Default::default()
        };

        publisher.publish(pesan).await.get().await?;
    }

    publisher.shutdown().await;
    println!("All events for user {} sent in order", pengguna_id);
    Ok(())
}

Subscription Filters — Only Receive Certain Messages #

async fn buat_subscription_dengan_filter(
    client: &Client,
    nama_subscription: &str,
    nama_topic: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    use google_cloud_pubsub::subscription::SubscriptionConfig;

    // Filter using the Common Expression Language (CEL)
    // Only receive messages with attribute event_type = "baru"
    let sub_config = SubscriptionConfig {
        filter: Some("attributes.event_type = \"baru\"".to_string()),
        ack_deadline_seconds: 30,
        ..Default::default()
    };

    client.create_subscription(
        nama_subscription,
        nama_topic,
        sub_config,
        None,
    ).await?;

    println!("Filter subscription created: only event_type = baru");
    Ok(())
}

Pub/Sub, SQS, and Kafka Comparison #

Aspect	Google Pub/Sub	Amazon SQS	Apache Kafka
Model	Topic-Subscription	Point-to-point Queue	Topic-Partition
Fan-out	Native (many subs per topic)	Via SNS	Different consumer groups
Ordering	Per ordering key	FIFO queues	Per partition
Replay	No (max 7 days)	No	Yes (offset)
Management	Fully managed	Fully managed	Self-managed or Confluent
Throughput	10 million msg/sec	Unlimited	Very high
Push mode	Yes (to HTTP endpoints)	No	No
Filtering	Yes (CEL expressions)	No	No
Ecosystem	GCP	AWS	Cloud-agnostic

Choose Google Pub/Sub when:
  ✓ Already in the GCP ecosystem (GKE, Cloud Run, Cloud Functions)
  ✓ Need fan-out: one message to many different consumers
  ✓ Push mode to Cloud Run or Cloud Functions
  ✓ Filter messages by attribute without extra code
  ✓ Native integration with BigQuery, Dataflow, GCS

Choose SQS when:
  ✓ Already in the AWS ecosystem
  ✓ Simple needs: task queues, one consumer per message

Choose Kafka when:
  ✓ Very high throughput and need event replay
  ✓ Not tied to a single cloud provider

Summary #

Every subscription gets a copy of all messages — unlike SQS where one message is consumed by only one consumer. Pub/Sub is closer to Kafka in this aspect: many subscriptions = many independent consumer groups.
Two consume modes: pull and streaming — pull for occasional batch processing, streaming pull (.receive()) for always-active consumers.
Ack is required to remove messages — unacked messages reappear after the ack deadline expires. Use nack() to deliberately return a message to the queue.
Ordering keys for per-entity order — messages with the same ordering key are guaranteed ordered within a subscription that enables enable_message_ordering. Use the entity ID as the key.
Attributes for metadata and filtering — send attributes with messages, create subscriptions with CEL filters to only receive certain messages without extra code.
Dead Letter Topics for failed messages — after max_delivery_attempts failed acks, the message is automatically forwarded to the DLT for investigation.
Automatic publisher batching — PublisherConfig.flush_interval and bundle_size control when batches are sent. Batching significantly reduces API call costs.
An emulator for local development — set PUBSUB_EMULATOR_HOST=localhost:8085 for testing without a GCP account.
Extend the ack deadline for long processes — modify_ack_deadline() prevents messages from reappearing in other consumers while still being processed.

← Previous: Amazon SQS Next: Redis →