Browse Apache Kafka Design Patterns & Streaming Architecture

Queue vs. Publish/Subscribe Models in Apache Kafka

November 25, 2024

Explore the Queue and Publish/Subscribe messaging models in Apache Kafka, understanding their characteristics, use cases, and implementation details for effective application in real-world scenarios.

On this page

4.1.1 Queue vs. Publish/Subscribe Models

In the realm of distributed systems and real-time data processing, understanding the nuances between different messaging models is crucial for designing efficient and scalable architectures. Apache Kafka, a leading platform for building real-time data pipelines and streaming applications, supports two primary messaging models: Queue and Publish/Subscribe. This section delves into these models, exploring their characteristics, use cases, and implementation details within Kafka.

Queue Model (Point-to-Point Messaging)

Intent

The Queue model, also known as point-to-point messaging, is designed to ensure that each message is processed by a single consumer. This model is particularly useful in scenarios where tasks need to be distributed among multiple consumers for load balancing and parallel processing.

Motivation

The Queue model is ideal for use cases where tasks or messages need to be processed exactly once by a single consumer. This ensures that work is evenly distributed across consumers, preventing duplication and ensuring efficient resource utilization.

Applicability

  • Task Distribution: When tasks need to be distributed among multiple workers for parallel processing.
  • Load Balancing: To balance the load across multiple consumers, ensuring that no single consumer is overwhelmed.
  • Order Processing: In scenarios where each order or transaction must be processed exactly once.

Structure

In Kafka, the Queue model is implemented using consumer groups. Each consumer group subscribes to a topic, and Kafka ensures that each message is delivered to only one consumer within the group.

    graph TD;
	    A["Producer"] -->|Send Messages| B["Kafka Topic"];
	    B -->|Distribute Messages| C["Consumer Group"];
	    C -->|Process Message| D["Consumer 1"];
	    C -->|Process Message| E["Consumer 2"];
	    C -->|Process Message| F["Consumer 3"];

Diagram: The Queue model in Kafka, where messages from a topic are distributed among consumers in a consumer group.

Participants

  • Producer: Sends messages to a Kafka topic.
  • Kafka Topic: Acts as a buffer, storing messages until they are consumed.
  • Consumer Group: A group of consumers that share the work of processing messages from a topic.
  • Consumers: Individual instances that process messages from the topic.

Collaborations

  • Producer to Topic: The producer sends messages to a Kafka topic.
  • Topic to Consumer Group: The topic distributes messages to consumers within a consumer group.
  • Consumer Group to Consumers: Each message is delivered to only one consumer within the group.

Consequences

  • Scalability: The Queue model allows for horizontal scaling by adding more consumers to the consumer group.
  • Fault Tolerance: If a consumer fails, Kafka can redistribute the messages to other consumers in the group.
  • Load Balancing: Ensures that messages are evenly distributed across consumers.

Implementation

  • Java:

     1import org.apache.kafka.clients.consumer.ConsumerConfig;
 2import org.apache.kafka.clients.consumer.KafkaConsumer;
 3import org.apache.kafka.clients.consumer.ConsumerRecords;
 4import org.apache.kafka.clients.consumer.ConsumerRecord;
 5
 6import java.util.Collections;
 7import java.util.Properties;
 8
 9public class QueueConsumer {
10    public static void main(String[] args) {
11        Properties props = new Properties();
12        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
13        props.put(ConsumerConfig.GROUP_ID_CONFIG, "queue-consumer-group");
14        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
15        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
16
17        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
18        consumer.subscribe(Collections.singletonList("queue-topic"));
19
20        while (true) {
21            ConsumerRecords<String, String> records = consumer.poll(100);
22            for (ConsumerRecord<String, String> record : records) {
23                System.out.printf("Consumed message: %s%n", record.value());
24            }
25        }
26    }
27}

  • Scala:

     1import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
 2import java.util.Properties
 3import scala.collection.JavaConverters._
 4
 5object QueueConsumer extends App {
 6  val props = new Properties()
 7  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
 8  props.put(ConsumerConfig.GROUP_ID_CONFIG, "queue-consumer-group")
 9  props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
10  props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
11
12  val consumer = new KafkaConsumer[String, String](props)
13  consumer.subscribe(List("queue-topic").asJava)
14
15  while (true) {
16    val records = consumer.poll(100).asScala
17    for (record <- records) {
18      println(s"Consumed message: ${record.value()}")
19    }
20  }
21}

  • Kotlin:

     1import org.apache.kafka.clients.consumer.ConsumerConfig
 2import org.apache.kafka.clients.consumer.KafkaConsumer
 3import java.util.*
 4
 5fun main() {
 6    val props = Properties()
 7    props[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = "localhost:9092"
 8    props[ConsumerConfig.GROUP_ID_CONFIG] = "queue-consumer-group"
 9    props[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = "org.apache.kafka.common.serialization.StringDeserializer"
10    props[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = "org.apache.kafka.common.serialization.StringDeserializer"
11
12    val consumer = KafkaConsumer<String, String>(props)
13    consumer.subscribe(listOf("queue-topic"))
14
15    while (true) {
16        val records = consumer.poll(100)
17        for (record in records) {
18            println("Consumed message: ${record.value()}")
19        }
20    }
21}

  • Clojure:

     1(require '[clojure.java.io :as io])
 2(import '[org.apache.kafka.clients.consumer KafkaConsumer ConsumerConfig]
 3        '[java.util Properties Collections])
 4
 5(defn create-consumer []
 6  (let [props (doto (Properties.)
 7                (.put ConsumerConfig/BOOTSTRAP_SERVERS_CONFIG "localhost:9092")
 8                (.put ConsumerConfig/GROUP_ID_CONFIG "queue-consumer-group")
 9                (.put ConsumerConfig/KEY_DESERIALIZER_CLASS_CONFIG "org.apache.kafka.common.serialization.StringDeserializer")
10                (.put ConsumerConfig/VALUE_DESERIALIZER_CLASS_CONFIG "org.apache.kafka.common.serialization.StringDeserializer"))]
11    (KafkaConsumer. props)))
12
13(defn consume-messages []
14  (let [consumer (create-consumer)]
15    (.subscribe consumer (Collections/singletonList "queue-topic"))
16    (while true
17      (let [records (.poll consumer 100)]
18        (doseq [record records]
19          (println "Consumed message:" (.value record)))))))
20
21(consume-messages)

Sample Use Cases

  • Financial Services: Processing transactions where each transaction must be handled exactly once.
  • E-commerce: Order processing systems where each order is processed by a single worker.
  • Manufacturing: Task distribution in assembly lines where each task is handled by a specific machine.
  • Competing Consumers: A pattern where multiple consumers compete to process messages from a queue, similar to the Queue model.
  • Load Balancing: Ensures that work is evenly distributed across consumers.

Publish/Subscribe Model

Intent

The Publish/Subscribe model allows messages to be broadcast to multiple consumers. This model is suitable for scenarios where the same message needs to be processed by multiple consumers, such as in event-driven architectures.

Motivation

The Publish/Subscribe model is ideal for use cases where messages need to be disseminated to multiple consumers, enabling them to react to events independently.

Applicability

  • Event-Driven Architectures: When multiple services need to react to the same event.
  • Real-Time Analytics: Broadcasting data to multiple analytics engines for processing.
  • Notification Systems: Sending alerts or notifications to multiple subscribers.

Structure

In Kafka, the Publish/Subscribe model is implemented using topics. Each consumer subscribes to a topic and receives all messages published to that topic.

    graph TD;
	    A["Producer"] -->|Broadcast Messages| B["Kafka Topic"];
	    B -->|Deliver Messages| C["Consumer 1"];
	    B -->|Deliver Messages| D["Consumer 2"];
	    B -->|Deliver Messages| E["Consumer 3"];

Diagram: The Publish/Subscribe model in Kafka, where messages from a topic are broadcast to all subscribed consumers.

Participants

  • Producer: Sends messages to a Kafka topic.
  • Kafka Topic: Acts as a channel, broadcasting messages to all subscribed consumers.
  • Consumers: Instances that subscribe to a topic and receive all messages.

Collaborations

  • Producer to Topic: The producer sends messages to a Kafka topic.
  • Topic to Consumers: The topic broadcasts messages to all subscribed consumers.

Consequences

  • Scalability: The Publish/Subscribe model allows for horizontal scaling by adding more consumers.
  • Decoupling: Consumers are decoupled from producers, allowing for independent scaling and development.
  • Flexibility: New consumers can be added without impacting existing consumers.

Implementation

  • Java:

     1import org.apache.kafka.clients.consumer.ConsumerConfig;
 2import org.apache.kafka.clients.consumer.KafkaConsumer;
 3import org.apache.kafka.clients.consumer.ConsumerRecords;
 4import org.apache.kafka.clients.consumer.ConsumerRecord;
 5
 6import java.util.Collections;
 7import java.util.Properties;
 8
 9public class PubSubConsumer {
10    public static void main(String[] args) {
11        Properties props = new Properties();
12        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
13        props.put(ConsumerConfig.GROUP_ID_CONFIG, "pubsub-consumer-group");
14        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
15        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
16
17        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
18        consumer.subscribe(Collections.singletonList("pubsub-topic"));
19
20        while (true) {
21            ConsumerRecords<String, String> records = consumer.poll(100);
22            for (ConsumerRecord<String, String> record : records) {
23                System.out.printf("Consumed message: %s%n", record.value());
24            }
25        }
26    }
27}

  • Scala:

     1import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
 2import java.util.Properties
 3import scala.collection.JavaConverters._
 4
 5object PubSubConsumer extends App {
 6  val props = new Properties()
 7  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
 8  props.put(ConsumerConfig.GROUP_ID_CONFIG, "pubsub-consumer-group")
 9  props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
10  props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
11
12  val consumer = new KafkaConsumer[String, String](props)
13  consumer.subscribe(List("pubsub-topic").asJava)
14
15  while (true) {
16    val records = consumer.poll(100).asScala
17    for (record <- records) {
18      println(s"Consumed message: ${record.value()}")
19    }
20  }
21}

  • Kotlin:

     1import org.apache.kafka.clients.consumer.ConsumerConfig
 2import org.apache.kafka.clients.consumer.KafkaConsumer
 3import java.util.*
 4
 5fun main() {
 6    val props = Properties()
 7    props[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = "localhost:9092"
 8    props[ConsumerConfig.GROUP_ID_CONFIG] = "pubsub-consumer-group"
 9    props[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = "org.apache.kafka.common.serialization.StringDeserializer"
10    props[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = "org.apache.kafka.common.serialization.StringDeserializer"
11
12    val consumer = KafkaConsumer<String, String>(props)
13    consumer.subscribe(listOf("pubsub-topic"))
14
15    while (true) {
16        val records = consumer.poll(100)
17        for (record in records) {
18            println("Consumed message: ${record.value()}")
19        }
20    }
21}

  • Clojure:

     1(require '[clojure.java.io :as io])
 2(import '[org.apache.kafka.clients.consumer KafkaConsumer ConsumerConfig]
 3        '[java.util Properties Collections])
 4
 5(defn create-consumer []
 6  (let [props (doto (Properties.)
 7                (.put ConsumerConfig/BOOTSTRAP_SERVERS_CONFIG "localhost:9092")
 8                (.put ConsumerConfig/GROUP_ID_CONFIG "pubsub-consumer-group")
 9                (.put ConsumerConfig/KEY_DESERIALIZER_CLASS_CONFIG "org.apache.kafka.common.serialization.StringDeserializer")
10                (.put ConsumerConfig/VALUE_DESERIALIZER_CLASS_CONFIG "org.apache.kafka.common.serialization.StringDeserializer"))]
11    (KafkaConsumer. props)))
12
13(defn consume-messages []
14  (let [consumer (create-consumer)]
15    (.subscribe consumer (Collections/singletonList "pubsub-topic"))
16    (while true
17      (let [records (.poll consumer 100)]
18        (doseq [record records]
19          (println "Consumed message:" (.value record)))))))
20
21(consume-messages)

Sample Use Cases

  • Social Media Platforms: Broadcasting updates or notifications to multiple users.
  • Stock Market: Distributing market data to multiple analytics engines.
  • IoT Systems: Sending sensor data to multiple processing units for analysis.
  • Event-Driven Architecture: A pattern where systems react to events, often implemented using the Publish/Subscribe model.
  • Observer Pattern: A design pattern where an object, known as the subject, maintains a list of its dependents, called observers, and notifies them of any state changes.

Performance Considerations and Scaling Implications

Queue Model

  • Performance: The Queue model can achieve high throughput by distributing messages across multiple consumers. However, it requires careful management of consumer offsets to ensure exactly-once processing.
  • Scaling: Adding more consumers to a consumer group can improve scalability, but it may require partitioning the topic to ensure that each consumer receives a balanced workload.

Publish/Subscribe Model

  • Performance: The Publish/Subscribe model can handle a large number of consumers, but each consumer receives a copy of every message, which can increase network and storage overhead.
  • Scaling: The model scales well with the addition of new consumers, but it may require careful management of topic partitions to ensure efficient message delivery.

Industries and Applications

Queue Model

  • Financial Services: Transaction processing systems where each transaction must be handled exactly once.
  • E-commerce: Order processing systems where each order is processed by a single worker.
  • Manufacturing: Task distribution in assembly lines where each task is handled by a specific machine.

Publish/Subscribe Model

  • Social Media Platforms: Broadcasting updates or notifications to multiple users.
  • Stock Market: Distributing market data to multiple analytics engines.
  • IoT Systems: Sending sensor data to multiple processing units for analysis.

Conclusion

Understanding the differences between the Queue and Publish/Subscribe models is essential for designing efficient and scalable messaging systems with Apache Kafka. By leveraging Kafka’s consumer groups and topics, developers can implement these models to meet the specific needs of their applications, whether it be task distribution, event-driven processing, or real-time analytics.

Test Your Knowledge: Queue vs. Publish/Subscribe Models in Kafka

Loading quiz…
Revised on Thursday, April 23, 2026
4.1.2 Hybrid Messaging Patterns