Google Cloud Pub/Sub Client Library

Warnings

• breaking Versions of `google-cloud-pubsub` from `2.35.0` and higher require Python 3.9 or newer. If you are using Python 3.7 or 3.8, you must pin the library version to `google-cloud-pubsub==2.34.0` or earlier.
  Fix: Upgrade your Python environment to 3.9+ or pin the library version: `pip install google-cloud-pubsub==2.34.0`.
• gotcha Instantiating multiple `PublisherClient` or `SubscriberClient` instances unnecessarily can lead to resource inefficiencies. These clients handle connection pooling and caching internally.
  Fix: For most applications, create a single `PublisherClient` and a single `SubscriberClient` instance per process and reuse them across operations to optimize resource utilization.
• gotcha Incorrectly configuring subscriber acknowledgment deadlines or flow control (prefetch settings) can cause 'stuck subscribers', messages being redelivered repeatedly (poison pill effect), or excessive resource consumption.
  Fix: Carefully tune your subscription's acknowledgment deadline to allow sufficient time for message processing. Implement appropriate flow control settings (`max_messages`, `max_bytes`) to prevent your application from being overwhelmed by messages. Always call `message.ack()` or `message.nack()` after processing.
• gotcha Pub/Sub guarantees at-least-once delivery, meaning a message might be delivered more than once in certain scenarios (e.g., subscriber restarts, ack deadline issues).
  Fix: Design your subscriber logic to be *idempotent*. Your message processing should produce the same result whether it's executed once or multiple times for the same message. Utilize unique business keys (like a transaction ID) and check against a fast-access store to prevent duplicate processing.
• gotcha Failing to capture and log the `message_id` returned by `publisher.publish().result()` can severely hinder debugging and traceability in production.
  Fix: Always store or log the `message_id` returned by the publish operation. This ID is the primary way to correlate a published message with its ingestion in Google Cloud Logs Explorer and track its lifecycle.
• gotcha When using the Pub/Sub emulator, incorrect `PUBSUB_EMULATOR_HOST` or `GOOGLE_CLOUD_PROJECT` environment variable settings can cause client libraries to connect to the production Pub/Sub service instead of the local emulator.
  Fix: Ensure `PUBSUB_EMULATOR_HOST` (e.g., `localhost:8085`) and `GOOGLE_CLOUD_PROJECT` (set to a placeholder like `my-project`) are correctly set in your environment *before* initializing `PublisherClient` or `SubscriberClient`.

Install

• pip install google-cloud-pubsub Install latest stable version

Imports

• PublisherClient

  from google.cloud import pubsub_v1
  publisher = pubsub_v1.PublisherClient()

  Older versions or incorrect top-level imports might not expose the v1 API clients directly. Always use `pubsub_v1` for current stable client access.
• SubscriberClient

  from google.cloud import pubsub_v1
  subscriber = pubsub_v1.SubscriberClient()

  Older versions or incorrect top-level imports might not expose the v1 API clients directly. Always use `pubsub_v1` for current stable client access.

Quickstart

This quickstart demonstrates how to publish a message to a Google Cloud Pub/Sub topic and then subscribe to and consume that message from a subscription. It handles topic and subscription creation if they don't exist and uses environment variables for project configuration.

import os
import time
from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1

project_id = os.environ.get('GOOGLE_CLOUD_PROJECT') or os.environ.get('GCP_PROJECT') or 'your-gcp-project-id'
topic_id = 'my-topic-id'
subscription_id = 'my-subscription-id'

if not project_id or project_id == 'your-gcp-project-id':
    raise ValueError("Please set the GOOGLE_CLOUD_PROJECT environment variable or replace 'your-gcp-project-id'.")

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

topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)

# Create topic if it doesn't exist
try:
    publisher.get_topic(request={"topic": topic_path})
    print(f"Topic {topic_path} already exists.")
except Exception:
    print(f"Creating topic {topic_path}...")
    publisher.create_topic(request={"name": topic_path})
    print(f"Topic {topic_path} created.")

# Create subscription if it doesn't exist
try:
    subscriber.get_subscription(request={"subscription": subscription_path})
    print(f"Subscription {subscription_path} already exists.")
except Exception:
    print(f"Creating subscription {subscription_path}...")
    subscriber.create_subscription(request={"name": subscription_path, "topic": topic_path})
    print(f"Subscription {subscription_path} created.")


# --- Publisher --- 
message_data = "Hello, Pub/Sub!"
print(f"Publishing message: '{message_data}' to {topic_path}")
future = publisher.publish(topic_path, message_data.encode('utf-8'))
message_id = future.result()
print(f"Published message with ID: {message_id}")

# --- Subscriber --- 
def callback(message: pubsub_v1.subscriber.message.Message):
    print(f"Received message: {message.data.decode('utf-8')}")
    print(f"Acknowledging message: {message.message_id}")
    message.ack()

print(f"Listening for messages on {subscription_path}...")
streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)

# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    try:
        # `subscribe` is non-blocking, so we must keep the main thread from exiting to allow it to run.
        streaming_pull_future.result(timeout=30) # Wait 30 seconds for messages
    except TimeoutError:
        streaming_pull_future.cancel()  # Trigger the shutdown.
        streaming_pull_future.result()  # Block until the shutdown is complete.
    except KeyboardInterrupt:
        streaming_pull_future.cancel()  # Trigger the shutdown.
        streaming_pull_future.result()  # Block until the shutdown is complete.

print("Finished listening for messages.")

# Clean up resources (optional)
# publisher.delete_topic(request={"topic": topic_path})
# subscriber.delete_subscription(request={"subscription": subscription_path})
# print(f"Topic {topic_id} and subscription {subscription_id} deleted.")