Kafka vs. RabbitMQ : Which Message Broker is Right for You?

What is RabbitMQ? 💁

Understanding RabbitMQ 🔍

Features of RabbitMQ

When to Use RabbitMQ ? 🐇

What is Kafka ? 🤷

Features of Kafka

When to Use Apache Kafka ?

The Difference Between RabbitMQ and Kafka

Architecture and Design Principles

• RabbitMQ follows a traditional message queue model, where messages are sent to a central broker (RabbitMQ server), which then routes them to the appropriate consumers.
• It relies on the Advanced Message Queuing Protocol (AMQP) and operates on a push-based approach, where the producer actively pushes messages to the broker for distribution.
• RabbitMQ's design emphasizes flexibility, extensibility, and support for various messaging patterns such as point-to-point, publish-subscribe, and request-reply.

• Kafka is designed as a distributed streaming platform that treats messages as events in an immutable log.
• It employs a distributed commit log architecture where messages are stored persistently and ordered across multiple brokers in a cluster.
• Kafka uses a pull-based model, where consumers actively fetch messages from the broker at their desired pace, enabling them to control their consumption rate.
• Kafka's design emphasizes high-throughput, fault tolerance, and real-time data streaming, making it well-suited for log aggregation, stream processing, and event-driven architectures.

Message Delivery Guarantees and Patterns

• RabbitMQ provides a strong emphasis on message reliability and guarantees.
• It ensures that messages are delivered exactly once using techniques like acknowledgments, message persistence, and automatic re-queueing in case of failures.
• RabbitMQ supports various messaging patterns such as point-to-point, publish-subscribe (via exchanges and queues), request-reply, and routing based on message headers.

• Kafka guarantees at-least-once message delivery semantics, where messages are persisted and replicated across multiple brokers for fault tolerance.
• It provides strong durability and fault tolerance by storing messages in its distributed commit log.
• Kafka excels in stream processing and real-time analytics, where it can handle high-throughput data streams efficiently while maintaining the order of events.

Use Cases and Typical Scenarios

• RabbitMQ is a suitable choice for applications that require reliable and guaranteed message delivery, especially in scenarios with complex routing and fanout patterns.
• It is well-suited for asynchronous communication, decoupling components, load balancing, and scenarios involving multiple consumers that need to process messages in parallel.

• Kafka is ideal for scenarios that involve high-throughput, real-time data streaming, log aggregation, and stream processing.
• It excels in event-driven architectures, real-time analytics, distributed systems, and scenarios where ordered event processing and fault tolerance are crucial.

Practical Demonstration: Building a Task Management App with Kafka and RabbitMQ

Set up the Backend

• Create a new directory for your project and navigate into it.
• Initialize a new Spring Boot project using Gradle by running the following command:

gradle init --type java-application

• Open the project in your preferred IDE. Update the build.gradle file to include the necessary dependencies. Here are the dependencies you'll need :

plugins {
    id 'org.springframework.boot' version '2.5.3'
    id 'io.spring.dependency-management' version '1.0.11.RELEASE'
}

dependencies {

    // Spring Boot Web Starter
    implementation 'org.springframework.boot:spring-boot-starter-web'

    // Lombok
    annotationProcessor 'org.projectlombok:lombok'

    // Kafka
    implementation 'org.springframework.kafka:spring-kafka'

    // RabbitMQ
    implementation 'org.springframework.boot:spring-boot-starter-amqp'

    // MapStruct
    annotationProcessor 'org.mapstruct:mapstruct-processor'

    // DevTools
    developmentOnly 'org.springframework.boot:spring-boot-devtools'

    // Test dependencies
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
}

Implementing the Backend

• By default, Spring Boot generates the main class file for its application. In our specific scenario, the main class file is named TaskBackendApplication.java, located within the src/main/java/com/example/taskbackend directory. You have the option to retain the code in its default state or modify it with the following code :

package com.example.taskbackend;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class TaskBackendApplication {

    public static void main(String[] args) {
        SpringApplication.run(TaskBackendApplication.class, args);
    }

    // You can add additional configurations or code here as needed.

}

• Navigate to the src/main/java/com/example/taskbackend directory and create a new directory named "controller". Create the TaskController.java file inside the “controller" directory. Update the TaskController.java file :

package com.example.taskbackend.controller;

import com.example.taskbackend.messaging.kafka.KafkaProducer;
import com.example.taskbackend.messaging.rabbitmq.RabbitMQProducer;
import com.example.taskbackend.model.Task;
import com.example.taskbackend.service.TaskService;
import org.springframework.web.bind.annotation.*;

import java.util.List;

@RestController
@RequestMapping("/api/tasks")
public class TaskController {

    private final TaskService taskService;
    private final KafkaProducer kafkaProducer;
    private final RabbitMQProducer rabbitMQProducer;

    public TaskController(TaskService taskService, KafkaProducer kafkaProducer, RabbitMQProducer rabbitMQProducer) {
        this.taskService = taskService;
        this.kafkaProducer = kafkaProducer;
        this.rabbitMQProducer = rabbitMQProducer;
    }

    @GetMapping
    public List<Task> getAllTasks() {
        return taskService.getAllTasks();
    }

    @PostMapping
    public Task createTask(@RequestBody Task task) {
        kafkaProducer.produce("New task created: " + task.getTitle());
        rabbitMQProducer.produce("New task created: " + task.getTitle());
        return taskService.createTask(task);
    }

    @PutMapping("/{id}")
    public Task updateTask(@PathVariable Long id, @RequestBody Task task) {
        return taskService.updateTask(id, task);
    }

    @DeleteMapping("/{id}")
    public void deleteTask(@PathVariable Long id) {
        taskService.deleteTask(id);
    }
}

• Inside project’s package directory, create a new directory “model”. Inside the newly created directory, create Task.java file.

package com.example.taskbackend.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Task {
    private Long id;
    private String title;
    private String description;
    private boolean completed;
}

• Open src/main/java/com/example/taskbackend directory and create a new directory named "service". Create the TaskService.java file inside the directory. Update the TaskService.java file with the codes below :

package com.example.taskbackend.service;

import com.example.taskbackend.model.Task;

import java.util.List;

public interface TaskService {
    List<Task> getAllTasks();
    Task createTask(Task task);
    Task updateTask(Long id, Task task);
    void deleteTask(Long id);
}

• Create a new directory called impl inside src/main/java/com/example/taskbackend. Within this newly created impl directory, add a file named TaskServiceImpl.java. This file will serve as the implementation class for the task service.

package com.example.taskbackend.service.impl;

import com.example.taskbackend.model.Task;
import com.example.taskbackend.service.TaskService;
import org.springframework.stereotype.Service;

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

@Service
public class TaskServiceImpl implements TaskService {

    private final Map<Long, Task> tasks = new HashMap<>();
    private Long nextId = 1L;

    @Override
    public List<Task> getAllTasks() {
        return new ArrayList<>(tasks.values());
    }

    @Override
    public Task createTask(Task task) {
        task.setId(nextId++);
        tasks.put(task.getId(), task);
        return task;
    }

    @Override
    public Task updateTask(Long id, Task task) {
        if (tasks.containsKey(id)) {
            task.setId(id);
            tasks.put(id, task);
            return task;
        }
        return null;
    }

    @Override
    public void deleteTask(Long id) {
        tasks.remove(id);
    }
}

• Proceed to generate a "messaging" directory inside src/main/java/com/example/taskbackend. Inside this "messaging" directory, establish two additional directories named "kafka" and "rabbitmq." Within the "kafka" directory, create two files: KafkaConsumer.java and KafkaProducer.java. Simultaneously, within the "rabbitmq" directory, craft two files as well: RabbitmqConsumer.java and RabbitmqProducer.java.

• Paste the codes below inside KafkaConsumer.java :

package com.example.taskbackend.messaging.kafka;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {

    @KafkaListener(topics = "task-topic", groupId = "task-group")
    public void consume(String message) {
        System.out.println("Received message from Kafka: " + message);
        // Add your custom logic here to process the message
    }
}

• Update KafkaProducer.java with the codes below :

package com.example.taskbackend.messaging.kafka;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class KafkaProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void produce(String message) {
        kafkaTemplate.send("task-topic", message);
        System.out.println("Produced message to Kafka: " + message);
    }
}

• Copy and paste the codes below into RabbitmqConsumer.java :

package com.example.taskbackend.messaging.rabbitmq;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class RabbitMQConsumer {

    @RabbitListener(queues = "task-queue")
    public void consume(String message) {
        System.out.println("Received message from RabbitMQ: " + message);
        // Add your custom logic here to process the message
    }
}

• Update RabbitmqProducer.java file with the codes below :

package com.example.taskbackend.messaging.rabbitmq;

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; // Add this import

@Configuration // Add this annotation
public class RabbitMQProducer {

    private final RabbitTemplate rabbitTemplate;

    public RabbitMQProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void produce(String message) {
        rabbitTemplate.convertAndSend("task-queue", message);
        System.out.println("Produced message to RabbitMQ: " + message);
    }

    @Bean
    public Queue taskQueue() {
        return new Queue("task-queue");
    }
}

• Finally, update your application.properties file with the code below :

spring.application.name=task-backend
server.port=8080
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.producer.bootstrap-servers=localhost:9092

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

• Run ./gradlew build to build the backend 😊

Setting up the Frontend

• Initialize a new React project with TypeScript by running the following command and prompts :

npm create vite@latest # create vite app
cd task-frontend # based on the project name you chose
npm install # to install react and other dependencies required by Vite
npm install axios bootstrap # to install Axios and Bootstrap

• Create the necessary directories for your frontend code :

• Inside src/components directory, create TaskList.tsx file. This will be the component for displaying the list of tasks. Update the file with codes below :

import React, { useEffect, useState } from 'react';
import axios from 'axios';

interface Task {
  id: number;
  title: string;
  description: string;
  completed: boolean;
}

const TaskList: React.FC = () => {
  const [tasks, setTasks] = useState<Task[]>([]);

  useEffect(() => {
    fetchTasks();
  }, []);

  const fetchTasks = async () => {
    try {
      const response = await axios.get<Task[]>('http://localhost:8080/api/tasks');
      setTasks(response.data);
    } catch (error) {
      console.error(error);
    }
  };

  return (
    <div>
      <h3>Task List</h3>
      {tasks.map((task) => (
        <div key={task.id}>
          <h4>{task.title}</h4>
          <p>{task.description}</p>
          <p>Completed: {task.completed ? 'Yes' : 'No'}</p>
        </div>
      ))}
    </div>
  );
};

export default TaskList;

• In that same directory, create TaskForm.tsx component. This will be the component for creating and updating tasks.

import React, { useState } from 'react';
import axios from 'axios';

interface TaskFormProps {
  onSubmit: () => void;
  sendToKafka: (message: string) => void;
  sendToRabbitMQ: (message: string) => void;
}

const TaskForm: React.FC<TaskFormProps> = ({ onSubmit, sendToKafka, sendToRabbitMQ }) => {
  const [title, setTitle] = useState('');
  const [description, setDescription] = useState('');
  const [error, setError] = useState('');

  const handleSubmit = async (event: React.FormEvent) => {
    event.preventDefault();

    if (!title || !description) {
      setError('Please enter both title and description.');
      return;
    }

    try {
      await axios.post('http://localhost:8080/api/tasks', { title, description, completed: false });
      setTitle('');
      setDescription('');
      setError('');
      onSubmit();
      sendToKafka(`New task created: ${title}`);
      sendToRabbitMQ(`New task created: ${title}`);
    } catch (error) {
      setError('An error occurred while creating the task.');
      console.error(error);
    }
  };

  return (
    <form onSubmit={handleSubmit}>
      <h3>Create Task</h3>
      {error && <p className="text-danger">{error}</p>}
      <div>
        <label htmlFor="title">Title</label>
        <br />
        <input
          type="text"
          id="title"
          value={title}
          onChange={(event) => setTitle(event.target.value)}
        />
      </div>
      <div>
        <label htmlFor="description">Description</label>
        <br />
        <textarea
          id="description"
          value={description}
          onChange={(event) => setDescription(event.target.value)}
        />
      </div>
      <button type="submit">Create</button>
    </form>
  );
};

export default TaskForm;

• Create services directory inside src directory. Within this directory, create TaskService.tsx file and paste the code below inside it :

import axios from 'axios';

interface Task {
  id: number;
  title: string;
  description: string;
  completed: boolean;
}

const taskService = {
  getAllTasks: async (): Promise<Task[]> => {
    try {
      const response = await axios.get<Task[]>('/api/tasks');
      return response.data;
    } catch (error) {
      console.error(error);
      return [];
    }
  },
  createTask: async (task: Task): Promise<Task | null> => {
    try {
      const response = await axios.post<Task>('/api/tasks', task);
      return response.data;
    } catch (error) {
      console.error(error);
      return null;
    }
  },
};

export default taskService;

• Update your App.tsx file with the codes below. This will be the main component for your application.

import React, { useEffect, useState } from 'react';
import axios from 'axios';
import TaskList from './components/TaskList';
import TaskForm from './components/TaskForm';

const App: React.FC = () => {
  const [refreshTasks, setRefreshTasks] = useState(false);

  useEffect(() => {
    setRefreshTasks(false);
  }, [refreshTasks]);

  const handleTaskSubmit = async () => {
    const newTask = {
      id: 0,
      title: 'New Task',
      description: 'Task description',
      completed: false,
    };
    await axios.post('http://localhost:8080/api/tasks', newTask);
    setRefreshTasks(true);
  };

  const sendToKafka = (message: string) => {
    // Use HTTP POST request to send message to Kafka
    axios.post('http://localhost:8080/api/tasks', { message })
      .then(() => {
        console.log('Sent message to Kafka:', message);
      })
      .catch((error) => {
        console.error('Failed to send message to Kafka:', error);
      });
  };

  const sendToRabbitMQ = (message: string) => {
    // Use HTTP POST request to send message to RabbitMQ
    axios.post('http://localhost:8080/api/tasks', { message })
      .then(() => {
        console.log('Sent message to RabbitMQ:', message);
      })
      .catch((error) => {
        console.error('Failed to send message to RabbitMQ:', error);
      });
  };

  return (
    <div className="container d-flex flex-column align-items-start justify-content-center vh-100">
      <h1 className="text-center">Task Management App</h1> <br/><br/>
      <div className="d-flex flex-column align-items-start">
        <div className="mb-3">
          <TaskForm onSubmit={handleTaskSubmit} sendToKafka={sendToKafka} sendToRabbitMQ={sendToRabbitMQ} />
        </div>
        <div>
          <TaskList key={refreshTasks.toString()} />
        </div>
      </div>
    </div>
  );
};

export default App;

• Finally, open main.tsx file and import Bootstrap :

import 'bootstrap/dist/css/bootstrap.css';

Configure, Docker, Kafka and RabbitMQ

FROM openjdk:20-jdk
WORKDIR /task-backend
COPY build/libs/task-backend-0.0.1-SNAPSHOT.jar task-backend-0.0.1-SNAPSHOT-plain.jar
EXPOSE 8080
ENTRYPOINT ["java", "-jar", "task-backend-0.0.1-SNAPSHOT-plain.jar"]

docker build -t backend .

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:6.2.0
    ports:
      - '2181:2181'
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
  kafka:
    image: confluentinc/cp-kafka:6.2.0
    ports:
      - '9092:9092'
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
  rabbitmq:
    image: rabbitmq:3.9.5
    ports:
      - '5672:5672'
      - '15672:15672'
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest
  backend:
    image: taskbackend:latest
    ports:
      - '8080:8080'

Conclusion

• the volume and velocity of data
• the need for message queuing and routing
• the availability of skilled personnel.