Google Pub/Sub #

Google Cloud Pub/Sub is Google Cloud’s fully managed distributed messaging service — designed to connect services that produce events with services that consume them, at global scale and low latency. The Pub/Sub (Publish-Subscribe) model differs from traditional queues: one message published to a Topic can be received by many different Subscriptions independently — each gets its own copy. This makes it ideal for event distribution, data pipelines, and microservice integration in the Google Cloud ecosystem. Understanding the difference between Pull vs Push subscriptions, how the acknowledgment deadline works, and configuring a Dead Letter Topic is key to a reliable implementation.

Pub/Sub Basic Concepts #

flowchart LR
    subgraph Publishers ["Publisher"]
        A["App A"]
        B["App B"]
    end

    subgraph TopicBox ["Topic"]
        T["orders"]
    end

    subgraph Subscriptions ["Subscription(s)"]
        SubP["payment-sub"]
        SubN["notif-sub"]
        SubA["audit-sub"]
    end

    subgraph Subscribers ["Subscriber"]
        P["Payment Service"]
        N["Notification Service"]
        C["Cloud Function / Run"]
    end

    A --> |publish| T
    B --> |publish| T
    
    T --> SubP
    T --> SubN
    T --> SubA

    SubP --> |pull| P
    SubN --> |pull| N
    SubA --> |push| C

    Note["Each subscription gets a copy of ALL messages"]
    T -.- Note

Important concepts:

Topic: message channel (publishers write here)
Subscription: a subscription to a topic; each sub has its own message queue
Pull: the subscriber actively requests messages (more common)
Push: Pub/Sub actively pushes messages to the subscriber’s HTTPS endpoint
Ack: the subscriber confirms a message is processed → removed from the sub
Nack: the subscriber rejects a message → redelivered immediately
Ack Deadline: the time (seconds) a subscriber has to ack before the message is redelivered

Installation #

pip install google-cloud-pubsub

Authentication #

Google Cloud supports several authentication methods. Application Default Credentials (ADC) is the recommended approach because it works automatically across different environments.

import os
from google.cloud import pubsub_v1

# ANTI-PATTERN: hardcoding the service account path in code
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/path/to/sa-key.json"  # ✗

# CORRECT way 1: ADC via an environment variable
# export GOOGLE_APPLICATION_CREDENTIALS="/path/to/sa-key.json"
# The library automatically reads from here

# CORRECT way 2: ADC via the gcloud CLI (for local development)
# gcloud auth application-default login

# CORRECT way 3: explicit Service Account (if you need credential isolation)
from google.oauth2 import service_account

credentials = service_account.Credentials.from_service_account_file(
    os.getenv("GOOGLE_APPLICATION_CREDENTIALS"),
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

publisher  = pubsub_v1.PublisherClient(credentials=credentials)
subscriber = pubsub_v1.SubscriberClient(credentials=credentials)

In GCP environments (GKE, Cloud Run, Cloud Functions, Compute Engine), use Workload Identity or a Service Account attached to the compute resource — no JSON file needed at all. ADC automatically fetches credentials from the GCP metadata server. This is the safest approach because credentials are rotated automatically and no secrets are stored in code or environment.

Creating Topics and Subscriptions #

import os
from google.cloud import pubsub_v1
from google.api_core.exceptions import AlreadyExists

PROJECT_ID = os.getenv("GCP_PROJECT_ID", "my-project")
publisher  = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()

def buat_topic(topic_id: str) -> str:
    topic_path = publisher.topic_path(PROJECT_ID, topic_id)
    try:
        topic = publisher.create_topic(request={"name": topic_path})
        print(f"Topic created: {topic.name}")
    except AlreadyExists:
        print(f"Topic already exists: {topic_path}")
    return topic_path

def buat_pull_subscription(
    topic_id:        str,
    subscription_id: str,
    ack_deadline:    int = 60,
    dlt_topic_id:    str = None
) -> str:
    topic_path = publisher.topic_path(PROJECT_ID, topic_id)
    sub_path   = subscriber.subscription_path(PROJECT_ID, subscription_id)

    request = {
        "name":                 sub_path,
        "topic":                topic_path,
        "ack_deadline_seconds": ack_deadline,
        "retry_policy": {
            "minimum_backoff": {"seconds": 10},
            "maximum_backoff": {"seconds": 600}
        }
    }

    if dlt_topic_id:
        dlt_path = publisher.topic_path(PROJECT_ID, dlt_topic_id)
        request["dead_letter_policy"] = {
            "dead_letter_topic":     dlt_path,
            "max_delivery_attempts": 5
        }

    try:
        sub = subscriber.create_subscription(request=request)
        print(f"Pull Subscription created: {sub.name}")
    except AlreadyExists:
        print(f"Subscription already exists: {sub_path}")

    return sub_path

def buat_push_subscription(
    topic_id:        str,
    subscription_id: str,
    push_endpoint:   str
) -> str:
    topic_path = publisher.topic_path(PROJECT_ID, topic_id)
    sub_path   = subscriber.subscription_path(PROJECT_ID, subscription_id)

    try:
        subscriber.create_subscription(request={
            "name":  sub_path,
            "topic": topic_path,
            "push_config": {"push_endpoint": push_endpoint},
            "ack_deadline_seconds": 30
        })
        print(f"Push Subscription created: {sub_path}")
    except AlreadyExists:
        print(f"Subscription already exists: {sub_path}")

    return sub_path

# Infrastructure setup
buat_topic("order-events")
buat_topic("order-events-dlt")
buat_pull_subscription(
    "order-events", "payment-service-sub",
    ack_deadline=60, dlt_topic_id="order-events-dlt"
)
buat_pull_subscription("order-events", "notification-service-sub", ack_deadline=30)

Publisher — Sending Messages #

import json
from datetime import datetime, timezone
from concurrent import futures

publisher  = pubsub_v1.PublisherClient()
TOPIC_PATH = publisher.topic_path(PROJECT_ID, "order-events")

def publish_pesan(payload: dict, attributes: dict = None) -> str:
    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")

    future = publisher.publish(
        TOPIC_PATH,
        data,
        **(attributes or {}),
        event_type=payload.get("event", "unknown"),
        source="order-service",
        version="1.0"
    )

    msg_id = future.result()
    print(f"Message sent, ID: {msg_id}")
    return msg_id

# Usage example
order_event = {
    "event":     "order.created",
    "order_id":  1001,
    "user_id":   42,
    "total":     18500000,
    "timestamp": datetime.now(timezone.utc).isoformat()
}
publish_pesan(order_event)

Batch Publishing #

def publish_batch(payloads: list[dict]) -> int:
    """Publish many messages asynchronously with automatic batching."""
    batch_settings = pubsub_v1.types.BatchSettings(
        max_bytes=1024 * 1024,
        max_latency=0.1,
        max_messages=100
    )
    pub   = pubsub_v1.PublisherClient(batch_settings=batch_settings)
    topic = pub.topic_path(PROJECT_ID, "order-events")

    publish_futures = []
    for payload in payloads:
        data   = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        future = pub.publish(topic, data, event_type=payload.get("event", "unknown"))
        publish_futures.append(future)

    sukses = 0
    for future in publish_futures:
        try:
            future.result()
            sukses += 1
        except Exception as e:
            print(f"Publish failed: {e}")

    print(f"{sukses}/{len(payloads)} messages published successfully.")
    return sukses

events = [{"event": "order.created", "order_id": i, "total": i * 50000} for i in range(1, 51)]
publish_batch(events)

Ordered Messages #

def publish_dengan_ordering(payload: dict, ordering_key: str) -> str:
    """Messages with the same ordering_key are guaranteed ordered."""
    pub = pubsub_v1.PublisherClient(
        publisher_options=pubsub_v1.types.PublisherOptions(
            enable_message_ordering=True
        )
    )
    topic = pub.topic_path(PROJECT_ID, "order-events-ordered")
    data  = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    return pub.publish(topic, data, ordering_key=ordering_key).result()

for status in ["created", "paid", "shipped", "delivered"]:
    publish_dengan_ordering(
        {"event": f"order.{status}", "order_id": 1001},
        ordering_key="user-42"
    )

Pull Subscriber — Receiving Messages #

Streaming Pull (Async) #

import signal

def proses_order(payload: dict) -> None:
    print(f"Processing order #{payload['order_id']} — Rp{payload['total']:,.0f}")

def jalankan_streaming_subscriber(subscription_id: str) -> None:
    sub      = pubsub_v1.SubscriberClient()
    sub_path = sub.subscription_path(PROJECT_ID, subscription_id)

    flow_control = pubsub_v1.types.FlowControl(
        max_messages=10,
        max_bytes=10 * 1024 * 1024
    )

    def callback(message: pubsub_v1.subscriber.message.Message) -> None:
        msg_id = message.message_id
        try:
            payload = json.loads(message.data.decode("utf-8"))
            print(f"Received [{msg_id}]: {message.attributes.get('event_type')}")

            proses_order(payload)
            message.ack()
            print(f"✓ ACK: {msg_id}")

        except json.JSONDecodeError:
            print(f"✗ Invalid message format: {msg_id}")
            message.ack()   # ack the broken message so it doesn't loop forever

        except Exception as e:
            print(f"✗ Error processing {msg_id}: {e}")
            # ANTI-PATTERN: message.ack() on error
            # message.ack()  # ✗ -- message lost!

            # CORRECT: nack so Pub/Sub redelivers immediately
            message.nack()  # ✓
            print(f"✗ NACK: {msg_id}")

    streaming_pull = sub.subscribe(sub_path, callback=callback, flow_control=flow_control)

    berjalan = True
    def handle_shutdown(signum, frame):
        nonlocal berjalan
        berjalan = False
        streaming_pull.cancel()

    signal.signal(signal.SIGINT,  handle_shutdown)
    signal.signal(signal.SIGTERM, handle_shutdown)

    print(f"Subscriber active on {sub_path}")
    try:
        streaming_pull.result()
    except Exception as e:
        if berjalan:
            print(f"Subscriber error: {e}")
    finally:
        sub.close()

jalankan_streaming_subscriber("payment-service-sub")

Synchronous Pull — for Batch Processing #

def pull_sinkron(subscription_id: str, maks_pesan: int = 10) -> list[dict]:
    sub      = pubsub_v1.SubscriberClient()
    sub_path = sub.subscription_path(PROJECT_ID, subscription_id)

    response = sub.pull(
        request={"subscription": sub_path, "max_messages": maks_pesan},
        timeout=30
    )

    if not response.received_messages:
        return []

    ack_ids = []
    hasil   = []

    for received in response.received_messages:
        msg = received.message
        try:
            payload = json.loads(msg.data.decode("utf-8"))
            hasil.append(payload)
            ack_ids.append(received.ack_id)
        except Exception as e:
            print(f"Parse error {msg.message_id}: {e}")

    if ack_ids:
        sub.acknowledge(request={"subscription": sub_path, "ack_ids": ack_ids})
        print(f"✓ Batch ACK: {len(ack_ids)} messages")

    sub.close()
    return hasil

Push Subscription — HTTPS Endpoint #

A push subscription is a mode where Pub/Sub actively pushes messages to your HTTPS endpoint — suitable for Cloud Run, Cloud Functions, or web applications.

import base64
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/pubsub/push", methods=["POST"])
def pubsub_push_handler():
    envelope = request.get_json()

    if not envelope or "message" not in envelope:
        return "Bad Request", 400

    msg      = envelope["message"]
    data_raw = base64.b64decode(msg.get("data", "")).decode("utf-8")

    try:
        payload = json.loads(data_raw)
        proses_order(payload)
        return jsonify({"status": "ok"}), 200   # 2xx = ACK

    except Exception as e:
        print(f"Error: {e}")
        return jsonify({"error": str(e)}), 500  # 5xx = NACK, Pub/Sub will retry

Extending the Ack Deadline #

import threading

def proses_dengan_extend_ack(message: pubsub_v1.subscriber.message.Message) -> None:
    """Periodically extend the ack deadline for long-running processing."""
    berhenti = threading.Event()

    def extend_loop():
        while not berhenti.wait(timeout=30):
            message.modify_ack_deadline(60)
            print(f"Ack deadline extended: {message.message_id}")

    thread = threading.Thread(target=extend_loop, daemon=True)
    thread.start()

    try:
        payload = json.loads(message.data.decode("utf-8"))
        import time
        for langkah in range(5):
            time.sleep(25)
            print(f"Step {langkah + 1}/5 done")
        message.ack()
    except Exception as e:
        message.nack()
    finally:
        berhenti.set()

Monitoring the Dead Letter Topic #

def pantau_dead_letter_topic(dlt_subscription_id: str) -> None:
    sub      = pubsub_v1.SubscriberClient()
    sub_path = sub.subscription_path(PROJECT_ID, dlt_subscription_id)

    response = sub.pull(
        request={"subscription": sub_path, "max_messages": 20},
        timeout=10
    )

    if not response.received_messages:
        print("Dead Letter Topic is empty.")
        sub.close()
        return

    print(f"⚠ {len(response.received_messages)} messages in the Dead Letter Topic:")
    ack_ids = []

    for received in response.received_messages:
        msg   = received.message
        attrs = msg.attributes
        try:
            payload = json.loads(msg.data.decode("utf-8"))
        except Exception:
            payload = msg.data.decode("utf-8")

        print(f"  ID: {msg.message_id}")
        print(f"  Attempted: {attrs.get('CloudPubSubDeadLetterSourceDeliveryCount', '?')}x")
        print(f"  From sub: {attrs.get('CloudPubSubDeadLetterSourceSubscription', '?')}")
        print(f"  Payload: {payload}\n")
        ack_ids.append(received.ack_id)

    if ack_ids:
        sub.acknowledge(request={"subscription": sub_path, "ack_ids": ack_ids})

    sub.close()

Summary #

Topics and Subscriptions are separate — each subscription gets an independent copy of every message; one topic can have many subscriptions for different services.
message.ack() on success, message.nack() on failure — ack() removes the message from the subscription; nack() asks Pub/Sub to redeliver immediately without waiting for the ack deadline to expire.
Flow control is required — set max_messages and max_bytes on the streaming subscriber so you aren’t flooded with more messages than you can process.
Dead Letter Topics (DLT) — configure dead_letter_policy on a subscription so messages that fail N times are automatically moved to the DLT for monitoring and debugging.
Push vs Pull — use Pull for full control over timing; use Push when you already have an HTTP endpoint and want Pub/Sub to actively push messages.
Ordering keys — use them to guarantee message order per entity (e.g., per user ID); only applies if the topic has message ordering enabled.
Batch publisher settings — configure max_messages, max_latency, and max_bytes to optimize publish throughput vs latency.
ADC and Workload Identity — on GCP, use Workload Identity or a Service Account attached to the compute resource; no need to store JSON credential files.
message.modify_ack_deadline() — for long processing, periodically extend the ack deadline so messages aren’t redelivered before you finish.
Retry policies on subscriptions — configure minimum_backoff and maximum_backoff to control the redelivery interval for failed messages.

← Previous: Amazon SQS Next: Redis →