In this lesson, we will explore the core concepts and operations of Google Cloud Pub/Sub, a messaging service that enables reliable, scalable, and asynchronous communication between different parts of your applications. Cloud Pub/Sub is designed around the concepts of topics and subscriptions, allowing you to publish messages to a topic and have them delivered to one or more subscribers. We will cover how to publish messages, configure subscriptions, and manage message delivery and acknowledgment. By the end of this lesson, you will understand how to use this messaging service effectively to build robust and decoupled systems.

To send information through Google Cloud Pub/Sub, you publish messages to a topic.

Publishing a message to a Cloud Pub/Sub topic:

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path('your-project-id', 'your-topic-name')

future = publisher.publish(topic_path, b'Hello world!')
print(f'Message published: {future.result()}')

Output:

Message published: 123456789012345678

You can also include custom attributes with your message:

future = publisher.publish(
    topic_path,
    b'Hello with attributes!',
    author='John Doe',
    weeks_on_job='10'
)
print(f'Message published: {future.result()}')

Output:

Message published: 123456789012345679

When you need to send multiple messages efficiently, Cloud Pub/Sub supports batch publishing. This allows you to publish several messages in a single request, reducing network overhead and improving throughput.

Batch publishing messages to a Cloud Pub/Sub topic:

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path('your-project-id', 'your-topic-name')

messages = [
    (b'This is the content of message 1', {'author': 'Jane Doe'}),
    (b'This is the content of message 2', {'author': 'John Smith'}),
    # Add more messages as needed
]

futures = []
for data, attrs in messages:
    future = publisher.publish(topic_path, data, **attrs)
    futures.append(future)

for future in futures:
    print(f'Message published: {future.result()}')

Output:

Message published: 123456789012345680
Message published: 123456789012345681

Batching is handled automatically by the client library, but you can also configure batching settings for more control if needed.

In Cloud Pub/Sub, messages published to a topic are delivered to all subscriptions attached to that topic. There are two main types of subscriptions: pull and push. With pull subscriptions, your application explicitly requests messages from the subscription. With push subscriptions, messages are automatically sent to a specified endpoint.

Pulling messages from a Cloud Pub/Sub subscription:

from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path('your-project-id', 'your-subscription-name')

response = subscriber.pull(
    request={
        "subscription": subscription_path,
        "max_messages": 10,
    }
)

for received_message in response.received_messages:
    print(f"Received: {received_message.message.data}")
    print(f"Message ID: {received_message.message.message_id}")
    print(f"Publish time: {received_message.message.publish_time}")
    if received_message.message.attributes:
        print(f"Attributes: {dict(received_message.message.attributes)}")
    # Acknowledge the message after processing
    subscriber.acknowledge(
        request={
            "subscription": subscription_path,
            "ack_ids": [received_message.ack_id],
        }
    )

Output:

Received: b'Hello world!'
Message ID: 123456789012345678
Publish time: 2024-01-15 10:30:45.123456+00:00
Received: b'Hello with attributes!'
Message ID: 123456789012345679
Publish time: 2024-01-15 10:30:45.234567+00:00
Attributes: {'author': 'John Doe', 'weeks_on_job': '10'}

With pull subscriptions, you control when and how many messages you retrieve, and you must acknowledge each message after processing.

In Cloud Pub/Sub, after a message is delivered to a subscriber, it must be acknowledged to confirm successful processing. If a message is not acknowledged within a certain period (the acknowledgment deadline), it will be redelivered. This ensures that messages are not lost if processing fails.

Acknowledging messages in Cloud Pub/Sub:

subscriber.acknowledge(
    request={
        "subscription": subscription_path,
        "ack_ids": [received_message.ack_id],
    }
)
print("Message acknowledged successfully")

Output:

Message acknowledged successfully

In this lesson, you learned how to use Google Cloud Pub/Sub to publish messages to topics, retrieve and process messages, and manage message acknowledgment. Cloud Pub/Sub enables scalable, asynchronous communication through topics and subscriptions. Understanding how to efficiently send, receive, and acknowledge messages is essential for building reliable and decoupled systems using this service. Keep practicing these operations to reinforce your knowledge and build robust cloud-based applications.