Welcome back! In this lesson, we will explore how to implement the fan-out messaging pattern using Google Cloud messaging services. The fan-out pattern allows a single publisher to send messages to multiple independent subscribers, enabling efficient and scalable communication between different parts of your application. By leveraging topics and multiple subscriptions, you can ensure that each subscriber receives a copy of every message published.
Initial setup
Consider a scenario with three services: Service A, Service B, and Service C. Service A needs to send updates to both Service B and Service C. Instead of sending messages directly to each service, we can create a topic that acts as a central channel. Both Service B and Service C can then subscribe to this topic, ensuring that they each receive all messages published by Service A.
Publishing to a Topic
Let's see how Service A can publish messages to a topic. We'll also include additional message attributes to provide more context to subscribers.
Python
from google.cloud import pubsub_v1# Initialize the publisher clientpublisher = pubsub_v1.PublisherClient()# Define the topic pathproject_id = "your-gcp-project-id"topic_id = "service-a-updates"topic_path = publisher.topic_path(project_id, topic_id)# Create the topic (if it doesn't exist)try: publisher.create_topic(request={"name": topic_path}) print(f"Topic created: {topic_path}")except Exception: print(f"Topic already exists: {topic_path}")# Publish a basic messagefuture = publisher.publish(topic_path, b"Updates for Services B and C")print(f"Published message ID: {future.result()}")# Publish a message with additional attributesfuture_advanced = publisher.publish( topic_path, b"News update", priority="high", department="IT")print(f"Published advanced message ID: {future_advanced.result()}")
In these examples, a message is published to the service-a-updates topic. The second message includes attributes such as priority and department, which can be used by subscribers to filter or process messages accordingly.
Join the 1M+ learners on CodeSignal
Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal
Creating Multiple Subscriptions for Fan-Out
To ensure that both Service B and Service C receive all messages from Service A, we create separate subscriptions to the same topic. Each subscription acts as an independent channel, allowing each service to process messages at its own pace.
from google.cloud import pubsub_v1# Initialize the subscriber clientsubscriber = pubsub_v1.SubscriberClient()# Define subscription pathssubscription_b_id = "service-b-sub"subscription_c_id = "service-c-sub"subscription_b_path = subscriber.subscription_path(project_id, subscription_b_id)subscription_c_path = subscriber.subscription_path(project_id, subscription_c_id)# Create subscriptions (if they don't exist)for subscription_path in [subscription_b_path, subscription_c_path]: try: subscriber.create_subscription( request={"name": subscription_path, "topic": topic_path} ) print(f"Subscription created: {subscription_path}") except Exception: print(f"Subscription already exists: {subscription_path}")print("Both subscriptions are ready to receive messages!")# Example: Pull messages for Service Bresponse = subscriber.pull( request={ "subscription": subscription_b_path, "max_messages": 10, })print(f"Pulled {len(response.received_messages)} messages for Service B:")for received_message in response.received_messages: message = received_message.message print(f"Message ID: {message.message_id}") print(f"Data: {message.data.decode('utf-8')}") print(f"Attributes: {dict(message.attributes)}") print(f"Publish time: {message.publish_time}") print("---") # Acknowledge the message subscriber.acknowledge( request={ "subscription": subscription_b_path, "ack_ids": [received_message.ack_id], } )print("All messages acknowledged for Service B")# Example: Pull messages for Service C to demonstrate fan-outresponse_c = subscriber.pull( request={ "subscription": subscription_c_path, "max_messages": 10, })print(f"\nPulled {len(response_c.received_messages)} messages for Service C:")for received_message in response_c.received_messages: message = received_message.message print(f"Message ID: {message.message_id}") print(f"Data: {message.data.decode('utf-8')}") print(f"Attributes: {dict(message.attributes)}") print("---") # Acknowledge the message subscriber.acknowledge( request={ "subscription": subscription_c_path, "ack_ids": [received_message.ack_id], } )print("All messages acknowledged for Service C")
Output:
Subscription created: projects/your-gcp-project-id/subscriptions/service-b-subSubscription created: projects/your-gcp-project-id/subscriptions/service-c-subBoth subscriptions are ready to receive messages!Pulled 2 messages for Service B:Message ID: 8234567890123456789Data: Updates for Services B and CAttributes: {}Publish time: 2024-01-15 10:30:45.123456+00:00---Message ID: 8234567890123456790Data: News updateAttributes: {'priority': 'high', 'department': 'IT'}Publish time: 2024-01-15 10:30:46.789012+00:00---All messages acknowledged for Service BPulled 2 messages for Service C:Message ID: 8234567890123456789Data: Updates for Services B and CAttributes: {}Publish time: 2024-01-15 10:30:45.123456+00:00---Message ID: 8234567890123456790Data: News updateAttributes: {'priority': 'high', 'department': 'IT'}Publish time: 2024-01-15 10:30:46.789012+00:00---All messages acknowledged for Service C
By creating multiple subscriptions to the same topic, each service receives its own copy of every message. This approach demonstrates the fan-out pattern, ensuring reliable and independent message delivery to all subscribers. Notice how both Service B and Service C received the same messages with identical message IDs, confirming that the fan-out pattern is working correctly.
Summary
In this lesson, we explored how to implement the fan-out messaging pattern using Google Cloud messaging services. By creating a topic and multiple subscriptions, a single publisher can efficiently distribute messages to multiple subscribers. This pattern enables scalable and reliable communication between different components of your application, allowing each subscriber to process messages independently. Experiment with creating additional topics and subscriptions to further customize your messaging architecture.
from google.cloud import pubsub_v1# Initialize the subscriber clientsubscriber = pubsub_v1.SubscriberClient()# Define subscription pathssubscription_b_id = "service-b-sub"subscription_c_id = "service-c-sub"subscription_b_path = subscriber.subscription_path(project_id, subscription_b_id)subscription_c_path = subscriber.subscription_path(project_id, subscription_c_id)# Create subscriptions (if they don't exist)for subscription_path in [subscription_b_path, subscription_c_path]: try: subscriber.create_subscription( request={"name": subscription_path, "topic": topic_path} ) print(f"Subscription created: {subscription_path}") except Exception: print(f"Subscription already exists: {subscription_path}")print("Both subscriptions are ready to receive messages!")# Example: Pull messages for Service Bresponse = subscriber.pull( request={ "subscription": subscription_b_path, "max_messages": 10, })print(f"Pulled {len(response.received_messages)} messages for Service B:")for received_message in response.received_messages: message = received_message.message print(f"Message ID: {message.message_id}") print(f"Data: {message.data.decode('utf-8')}") print(f"Attributes: {dict(message.attributes)}") print(f"Publish time: {message.publish_time}") print("---") # Acknowledge the message subscriber.acknowledge( request={ "subscription": subscription_b_path, "ack_ids": [received_message.ack_id], } )print("All messages acknowledged for Service B")# Example: Pull messages for Service C to demonstrate fan-outresponse_c = subscriber.pull( request={ "subscription": subscription_c_path, "max_messages": 10, })print(f"\nPulled {len(response_c.received_messages)} messages for Service C:")for received_message in response_c.received_messages: message = received_message.message print(f"Message ID: {message.message_id}") print(f"Data: {message.data.decode('utf-8')}") print(f"Attributes: {dict(message.attributes)}") print("---") # Acknowledge the message subscriber.acknowledge( request={ "subscription": subscription_c_path, "ack_ids": [received_message.ack_id], } )print("All messages acknowledged for Service C")
text
Subscription created: projects/your-gcp-project-id/subscriptions/service-b-subSubscription created: projects/your-gcp-project-id/subscriptions/service-c-subBoth subscriptions are ready to receive messages!Pulled 2 messages for Service B:Message ID: 8234567890123456789Data: Updates for Services B and CAttributes: {}Publish time: 2024-01-15 10:30:45.123456+00:00---Message ID: 8234567890123456790Data: News updateAttributes: {'priority': 'high', 'department': 'IT'}Publish time: 2024-01-15 10:30:46.789012+00:00---All messages acknowledged for Service BPulled 2 messages for Service C:Message ID: 8234567890123456789Data: Updates for Services B and CAttributes: {}Publish time: 2024-01-15 10:30:45.123456+00:00---Message ID: 8234567890123456790Data: News updateAttributes: {'priority': 'high', 'department': 'IT'}Publish time: 2024-01-15 10:30:46.789012+00:00---All messages acknowledged for Service C
