Implementing a cache with distributed event streaming using Apache Kafka and HashMap in Java

23 Oct 2023

In this blog post, we will explore how to implement a cache using distributed event streaming with Apache Kafka and a HashMap in Java. Caching is a technique used to improve the performance of an application by storing frequently accessed data in memory.

Introduction
Setting Up Apache Kafka
Creating a Cache with HashMap
Event Streaming with Apache Kafka
Building the Cache Service
Conclusion
References

1. Introduction

Caching data can greatly reduce the response time of an application, especially when the data is expensive to retrieve or compute. In a distributed system, maintaining a cache can be challenging, as multiple instances of an application need access to the same cache.

2. Setting Up Apache Kafka

To get started, you need to set up Apache Kafka on your machine. Follow the official Apache Kafka documentation to install and configure Kafka.

3. Creating a Cache with HashMap

First, let’s create a simple cache using a HashMap in Java. The HashMap will store key-value pairs, where the key represents the data to be cached, and the value represents the cached data.

import java.util.HashMap;

public class Cache {
    private static HashMap<String, Object> cache = new HashMap<>();

    public static void put(String key, Object value) {
        cache.put(key, value);
    }

    public static Object get(String key) {
        return cache.get(key);
    }

    public static boolean contains(String key) {
        return cache.containsKey(key);
    }

    public static void remove(String key) {
        cache.remove(key);
    }
}

4. Event Streaming with Apache Kafka

With the cache implemented, we can now start with the event streaming part using Apache Kafka. Apache Kafka is a distributed event streaming platform that allows us to publish and consume events in real-time.

To use Kafka in Java, you will need to add the Kafka client dependency to your project. You can do this by adding the following Maven dependency to your pom.xml file:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

Next, you need to create a Kafka producer and consumer to publish and consume events respectively. Here’s a simplified example:

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaUtils {
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String TOPIC = "cache-events";

    public static void produceEvent(String key, Object value) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<>(props);

        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, key, value.toString());

        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("Event published successfully to topic: " + metadata.topic());
            } else {
                System.err.println("Error publishing event to topic: " + exception.getMessage());
            }
        });

        producer.close();
    }

    public static void consumeEvents() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "cache-consumer");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(List.of(TOPIC));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                String key = record.key();
                String value = record.value();

                // Update the cache with the received event
                Cache.put(key, value);
            }
        }
    }
}

5. Building the Cache Service

With the cache and Kafka utilities in place, we can now build our cache service. The cache service will have methods to interact with the cache and leverage Kafka for event streaming.

import java.util.Scanner;

public class CacheService {
    public static void main(String[] args) {
        // Start consuming events from Kafka
        new Thread(KafkaUtils::consumeEvents).start();

        Scanner scanner = new Scanner(System.in);

        while (true) {
            System.out.print("Enter a key: ");
            String key = scanner.nextLine();

            if (Cache.contains(key)) {
                System.out.println("Value from cache: " + Cache.get(key));
            } else {
                System.out.print("Enter a value: ");
                String value = scanner.nextLine();

                Cache.put(key, value);

                // Publish the event to Kafka
                KafkaUtils.produceEvent(key, value);

                System.out.println("Value saved to cache");
            }
        }
    }
}

6. Conclusion

In this blog post, we learned how to implement a cache using distributed event streaming with Apache Kafka and a HashMap in Java. Caching can greatly improve the performance of an application, and using a distributed cache ensures that all instances of the application have access to the same cache.

By leveraging Apache Kafka, we can ensure that cache updates are propagated to all instances of the application in real-time. This allows for a scalable and reliable cache solution.

Table of Contents