Design Pattern: Publisher-Subscriber

By Amr Saafan Engineering, Software Design, Technical Design Pattern, callbacks, Java, JavaScript, Kafka, Message Brokers, Microservices, Publisher, PubSub, python, rabbitmq, Real-Time, subscriber

The Publisher/Subscriber Pattern, commonly referred to as Pub/Sub, is a powerful design pattern that is essential in modern software development, especially for implementing event-driven architecture and decoupling system components. This article delves deeply into the Pub/Sub pattern, exploring its principles, benefits, and practical implementation with extensive code examples to ensure a thorough understanding.

Introduction to the Publisher/Subscriber Pattern

A communications paradigm known as the Publisher/Subscriber pattern involves senders (publishers) broadcasting messages without knowing who the recipients are (subscribers). Without knowing the publishers, subscribers indicate interest in one or more themes, and they get interesting communications. The system’s flexibility and scalability are improved by this separation of publishers and subscribers.

Key Concepts:

Publisher: The component that sends messages.
Subscriber: The component that receives messages.
Message: The data being transmitted.
Channel/Topic: The medium through which messages are sent.

The Publisher/Subscriber pattern is particularly useful in applications where components need to communicate asynchronously and independently, such as in microservices architecture, real-time systems, and distributed systems.

Benefits of the Publisher/Subscriber Pattern

Decoupling: Publishers and subscribers are independent of each other, allowing for more flexible and maintainable code.
Scalability: The pattern supports adding more publishers or subscribers without significant changes to the system.
Flexibility: Subscribers can dynamically subscribe or unsubscribe from topics, allowing for real-time changes in behavior.
Asynchronous Communication: Messages can be sent and received without blocking the execution of the publisher or subscriber.

Implementation of the Publisher/Subscriber Pattern

Let’s explore how to implement the Publisher/Subscriber pattern in different programming languages, including JavaScript, Python, and Java.

JavaScript Implementation

In JavaScript, the Publisher/Subscriber pattern can be implemented using simple objects and arrays. Here’s a basic implementation:

class PubSub {
    constructor() {
        this.subscribers = {};
    }

    subscribe(event, callback) {
        if (!this.subscribers[event]) {
            this.subscribers[event] = [];
        }
        this.subscribers[event].push(callback);
    }

    unsubscribe(event, callback) {
        if (!this.subscribers[event]) return;

        this.subscribers[event] = this.subscribers[event].filter(subscriber => subscriber !== callback);
    }

    publish(event, data) {
        if (!this.subscribers[event]) return;

        this.subscribers[event].forEach(callback => callback(data));
    }
}

// Usage
const pubSub = new PubSub();

const onUserAdded = (user) => {
    console.log(`User added: ${user.name}`);
};

pubSub.subscribe('userAdded', onUserAdded);
pubSub.publish('userAdded', { name: 'John Doe' });

In this example, we define a PubSub class with methods to subscribe, unsubscribe, and publish events. Subscribers register their interest in specific events and are notified when those events occur.

Python Implementation

In Python, the Publisher/Subscriber pattern can be implemented using dictionaries and lists:

class PubSub:
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, event, callback):
        if event not in self.subscribers:
            self.subscribers[event] = []
        self.subscribers[event].append(callback)

    def unsubscribe(self, event, callback):
        if event in self.subscribers:
            self.subscribers[event].remove(callback)

    def publish(self, event, data):
        if event in self.subscribers:
            for callback in self.subscribers[event]:
                callback(data)

# Usage
pubsub = PubSub()

def on_user_added(user):
    print(f"User added: {user['name']}")

pubsub.subscribe('userAdded', on_user_added)
pubsub.publish('userAdded', {'name': 'Jane Doe'})

This Python implementation mirrors the JavaScript version, using methods to handle subscribing, unsubscribing, and publishing events.

Java Implementation

In Java, the Publisher/Subscriber pattern can be implemented using interfaces and classes:

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

interface Subscriber {
    void update(String event, Object data);
}

class PubSub {
    private Map<String, List<Subscriber>> subscribers = new HashMap<>();

    public void subscribe(String event, Subscriber subscriber) {
        subscribers.computeIfAbsent(event, k -> new ArrayList<>()).add(subscriber);
    }

    public void unsubscribe(String event, Subscriber subscriber) {
        List<Subscriber> subs = subscribers.get(event);
        if (subs != null) {
            subs.remove(subscriber);
        }
    }

    public void publish(String event, Object data) {
        List<Subscriber> subs = subscribers.get(event);
        if (subs != null) {
            for (Subscriber subscriber : subs) {
                subscriber.update(event, data);
            }
        }
    }
}

// Usage
class UserAddedSubscriber implements Subscriber {
    public void update(String event, Object data) {
        System.out.println("User added: " + ((Map)data).get("name"));
    }
}

public class PubSubExample {
    public static void main(String[] args) {
        PubSub pubSub = new PubSub();
        Subscriber userAddedSubscriber = new UserAddedSubscriber();

        pubSub.subscribe("userAdded", userAddedSubscriber);
        pubSub.publish("userAdded", Map.of("name", "John Doe"));
    }
}

In this Java example, we use an interface for subscribers and a PubSub class to manage subscriptions and events. Subscribers implement the Subscriber interface and define their update method to handle incoming messages.

Advanced Usage and Considerations

Handling Complex Scenarios

In real-world applications, the Publisher/Subscriber pattern can be extended to handle more complex scenarios, such as:

Message Filtering: Allowing subscribers to filter messages based on specific criteria.
Persistent Messaging: Storing messages in a persistent store to ensure delivery even if subscribers are temporarily unavailable.
Message Brokers: Using message brokers like RabbitMQ, Kafka, or Redis to manage and route messages.

Message Filtering Example

Here’s an example of implementing message filtering in JavaScript:

class PubSub {
    constructor() {
        this.subscribers = {};
    }

    subscribe(event, callback, filter = () => true) {
        if (!this.subscribers[event]) {
            this.subscribers[event] = [];
        }
        this.subscribers[event].push({ callback, filter });
    }

    unsubscribe(event, callback) {
        if (!this.subscribers[event]) return;

        this.subscribers[event] = this.subscribers[event].filter(subscriber => subscriber.callback !== callback);
    }

    publish(event, data) {
        if (!this.subscribers[event]) return;

        this.subscribers[event].forEach(subscriber => {
            if (subscriber.filter(data)) {
                subscriber.callback(data);
            }
        });
    }
}

// Usage
const pubSub = new PubSub();

const onAdultUserAdded = (user) => {
    console.log(`Adult user added: ${user.name}`);
};

pubSub.subscribe('userAdded', onAdultUserAdded, user => user.age >= 18);
pubSub.publish('userAdded', { name: 'John Doe', age: 20 });
pubSub.publish('userAdded', { name: 'Jane Doe', age: 17 });

In this example, subscribers can provide a filter function that determines whether they should receive a specific message.

Integrating Publisher/Subscriber with Message Brokers

Using a message broker can help manage complexity and improve reliability and scalability. Here’s an example using Node.js with RabbitMQ:

const amqp = require('amqplib/callback_api');

class PubSub {
    constructor() {
        this.connection = null;
        this.channel = null;
    }

    async connect() {
        return new Promise((resolve, reject) => {
            amqp.connect('amqp://localhost', (error, connection) => {
                if (error) {
                    reject(error);
                    return;
                }
                this.connection = connection;
                connection.createChannel((error, channel) => {
                    if (error) {
                        reject(error);
                        return;
                    }
                    this.channel = channel;
                    resolve();
                });
            });
        });
    }

    publish(queue, message) {
        this.channel.assertQueue(queue, { durable: false });
        this.channel.sendToQueue(queue, Buffer.from(message));
    }

    subscribe(queue, callback) {
        this.channel.assertQueue(queue, { durable: false });
        this.channel.consume(queue, (msg) => {
            callback(msg.content.toString());
        }, { noAck: true });
    }
}

// Usage
(async () => {
    const pubSub = new PubSub();
    await pubSub.connect();

    pubSub.subscribe('userAdded', (message) => {
        console.log(`Received: ${message}`);
    });

    pubSub.publish('userAdded', JSON.stringify({ name: 'John Doe', age: 20 }));
})();

This example demonstrates how to use RabbitMQ to implement the Publisher/Subscriber pattern, leveraging a message broker for message routing and delivery.

Real-World Applications

The Publisher/Subscriber pattern is widely used in various real-world applications:

Microservices Architecture: Facilitating communication between microservices without tight coupling.
Real-Time Applications: Implementing real-time updates in applications like chat systems, notifications, and live feeds.
Event Sourcing: Capturing state changes as a sequence of events, useful in financial systems and auditing.

Microservices Example

In a microservices architecture, the Publisher/

Subscriber pattern can be used to decouple services and enable them to communicate asynchronously. Here’s a simplified example:

User Service (Publisher):

import pika

def publish_message(queue, message):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue=queue)
    channel.basic_publish(exchange='', routing_key=queue, body=message)
    connection.close()

# Usage
publish_message('userAdded', 'User John Doe added')

Notification Service (Subscriber):

import pika

def callback(ch, method, properties, body):
    print(f"Received: {body}")

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='userAdded')
channel.basic_consume(queue='userAdded', on_message_callback=callback, auto_ack=True)
print('Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

In this example, the User Service publishes a message when a new user is added, and the Notification Service subscribes to receive these messages and send notifications.

Best Practices and Considerations

Error Handling: Implement robust error handling to ensure the system can recover from failures.
Performance: Optimize the performance of the messaging system, especially in high-throughput scenarios.
Security: Ensure secure transmission of messages, particularly in distributed systems.

Error Handling Example

Here’s an example of adding error handling to the JavaScript Pub/Sub implementation:

class PubSub {
    constructor() {
        this.subscribers = {};
    }

    subscribe(event, callback) {
        if (!this.subscribers[event]) {
            this.subscribers[event] = [];
        }
        this.subscribers[event].push(callback);
    }

    unsubscribe(event, callback) {
        if (!this.subscribers[event]) return;

        this.subscribers[event] = this.subscribers[event].filter(subscriber => subscriber !== callback);
    }

    publish(event, data) {
        if (!this.subscribers[event]) return;

        this.subscribers[event].forEach(callback => {
            try {
                callback(data);
            } catch (error) {
                console.error(`Error in subscriber callback: ${error}`);
            }
        });
    }
}

// Usage
const pubSub = new PubSub();

const onUserAdded = (user) => {
    if (!user.name) {
        throw new Error('User name is required');
    }
    console.log(`User added: ${user.name}`);
};

pubSub.subscribe('userAdded', onUserAdded);
pubSub.publish('userAdded', { name: 'John Doe' });
pubSub.publish('userAdded', {});  // This will trigger an error

In this example, we add error handling within the publish method to catch and log errors from subscriber callbacks.

Conclusion

One basic design pattern that offers a reliable way to separate components and create event-driven systems is the Publisher/Subscriber paradigm. Build scalable, stable, and adaptable systems by utilizing the Pub/Sub pattern by comprehending its concepts, advantages, and application. When it comes to distributed systems, real-time apps, and microservices, the Publisher/Subscriber paradigm is a crucial component of every designer’s toolbox.

Cookie	Duration	Description
cookielawinfo-checkbox-analytics	11 months	This cookie is set by GDPR Cookie Consent plugin. The cookie is used to store the user consent for the cookies in the category "Analytics".
cookielawinfo-checkbox-functional	11 months	The cookie is set by GDPR cookie consent to record the user consent for the cookies in the category "Functional".
cookielawinfo-checkbox-necessary	11 months	This cookie is set by GDPR Cookie Consent plugin. The cookies is used to store the user consent for the cookies in the category "Necessary".
cookielawinfo-checkbox-others	11 months	This cookie is set by GDPR Cookie Consent plugin. The cookie is used to store the user consent for the cookies in the category "Other.
cookielawinfo-checkbox-performance	11 months	This cookie is set by GDPR Cookie Consent plugin. The cookie is used to store the user consent for the cookies in the category "Performance".
viewed_cookie_policy	11 months	The cookie is set by the GDPR Cookie Consent plugin and is used to store whether or not user has consented to the use of cookies. It does not store any personal data.

Design Pattern: Publisher-Subscriber