Showing posts with label Java. Show all posts
Showing posts with label Java. Show all posts

Saturday, 28 November 2020

Write a concurrent LRU Cache using ReadWriteLock to allow parallel reads

 

github: https://github.com/prasune/Algorithms/tree/master/src/main/java/com/test/algorithm/cache

This LRUCache is implemented using ReadWriteLock allowing parallel reads where as only a single write can happen at a time. But caches are meant for read with less write and this approach would help in parallel reads for small scale of data.

This approach has a major disadvantage with respect to time taken for searching an entry if the scale of data to be retained is huge. This is because the ConcurrentLinkedQueue needs to be scanned to re-add after picking up the data every time. For large scale data it is better to use the synchronized LinkedHashMap approach.


package com.test.algorithm.cache;

import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class LRUCacheWithReadWriteLock<K, V> {

private final int limit;
private final Map<K, V> internalCache;
private final Queue<K> trackingQueue;

private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

public LRUCacheWithReadWriteLock(int limit){
internalCache = new ConcurrentHashMap<>();
trackingQueue = new ConcurrentLinkedQueue<>();
this.limit = limit;
}

public V get(K key) {
this.readWriteLock.readLock().lock();
try {
V value = internalCache.get(key);
if (value != null) {
if (trackingQueue.remove(key)) {
trackingQueue.add(key);
}
}
return value;
} finally {
this.readWriteLock.readLock().unlock();
}
}

public void put(K key, V value) {
this.readWriteLock.writeLock();
try {
if (internalCache.containsKey(key)) {
trackingQueue.remove(key);
}
if (trackingQueue.size() == limit) {
K expiredKey = trackingQueue.poll();
internalCache.remove(expiredKey);
}
internalCache.put(key, value);
trackingQueue.add(key);
} finally {
this.readWriteLock.writeLock().unlock();
}
}

public static void main(String arg[]){
LRUCacheWithReadWriteLock<Integer, String> lruCache = new LRUCacheWithReadWriteLock<>(4);

lruCache.put(1, "Object1");
lruCache.put(2, "Object2");
lruCache.put(3, "Object3");
lruCache.get(1);
lruCache.put(4, "Object4");
System.out.println(lruCache);
lruCache.put(5, "Object5");
lruCache.get(3);
lruCache.put(6, "Object6");
System.out.println(lruCache);
lruCache.get(4);
lruCache.put(7, "Object7");
lruCache.put(8, "Object8");
System.out.println(lruCache);
}

@Override
public String toString() {
return "LRUCacheWithReadWriteLock{" +
"internalCache=" + internalCache +
'}';
}
}

Write a concurrent LRU Cache using LinkedHashMap

 

github: https://github.com/prasune/Algorithms/tree/master/src/main/java/com/test/algorithm/cache

For handling concurrency, we will have to synchronize the methods of LinkedHashMap being used as LRU Cache.

For better performance, LRU Cache can be implemented using ReadWriteLock, ConcurrentLinkedQueue and a ConcurrentHashMap - refer the implementation

package com.test.algorithm.cache;

import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;

public class LRUCacheWithLinkedHashMap<K, V> {

private final Map<K, V> internalCache;

public LRUCacheWithLinkedHashMap(int limit){
internalCache = Collections.synchronizedMap(new LinkedHashMap<K, V>(limit, 0.75f, true) {
@Override
protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
return size() > limit;
}
});
}

public V get(K key) {
return internalCache.get(key);
}

public void put(K key, V value) {
internalCache.put(key, value);
}

public V remove(K key) {
return internalCache.remove(key);
}

public static void main(String arg[]){
LRUCacheWithLinkedHashMap<Integer, String> lruCache = new LRUCacheWithLinkedHashMap<>(4);

lruCache.put(1, "Object1");
lruCache.put(2, "Object2");
lruCache.put(3, "Object3");
lruCache.get(1);
lruCache.put(4, "Object4");
System.out.println(lruCache);
lruCache.put(5, "Object5");
lruCache.get(3);
lruCache.put(6, "Object6");
System.out.println(lruCache);
lruCache.get(4);
lruCache.put(7, "Object7");
lruCache.put(8, "Object8");
System.out.println(lruCache);
}

@Override
public String toString() {
return "LRUCacheWithLinkedHashMap{" +
"internalCache=" + internalCache +
'}';
}
}

Wednesday, 25 November 2020

Write a program to print top N trending hashtags

 You are given a set of hashtags and the number of times those have occured. 

Write a java program to print top N trending hashtags from the list of values.

Solution: The best suited data structure for figuring out top N elements is a heap - PriorityQueue is the java implementation for heap.

github: https://github.com/prasune/Algorithms/tree/master/src/main/java/com/test/algorithm/heap

package com.test.algorithm.heap;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;

public class TopNHashtags {

public static void main(String[] args) {
int n = 3;
List<HashTagCount> hashTagList = new ArrayList<>();
hashTagList.add(new HashTagCount("#sometag", 10));
hashTagList.add(new HashTagCount("#sometag1", 1000));
hashTagList.add(new HashTagCount("#sometag2", 900));
hashTagList.add(new HashTagCount("#sometag3", 100));
hashTagList.add(new HashTagCount("#sometag4", 50));
hashTagList.add(new HashTagCount("#sometag", 10));

printTopNTrendingHashTags(n, hashTagList);
}

private static void printTopNTrendingHashTags(int n, List<HashTagCount> hashTagList) {
PriorityQueue<HashTagCount> topTrendingHashTags =
new PriorityQueue<>(Comparator.comparing(HashTagCount::getCount).reversed());
topTrendingHashTags.addAll(hashTagList);
for (int i = 0; i< n; i++) {
System.out.println(topTrendingHashTags.poll());
}
}
}


package com.test.algorithm.heap;

import java.util.Objects;

public class HashTagCount {

private String hashTag;

private long count;

public HashTagCount(String hashTag, long count) {
this.hashTag = hashTag;
this.count = count;
}

public String getHashTag() {
return hashTag;
}

public void setHashTag(String hashTag) {
this.hashTag = hashTag;
}

public long getCount() {
return count;
}

public void setCount(long count) {
this.count = count;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
HashTagCount that = (HashTagCount) o;
return hashTag.equals(that.hashTag);
}

@Override
public int hashCode() {
return Objects.hash(hashTag);
}

@Override
public String toString() {
return "TweetCount{" +
"hashTag='" + hashTag + '\'' +
", count=" + count +
'}';
}
}

Friday, 23 October 2020

JDBI 3 - How to enable sql logging

 

Sometimes we would require to enable printing of sqls along with the execution time while debugging the issues or doing psr testing. 

Following are the steps to enable sql logging with time taken to execute in JDBI3:

1. Add the following code on injected Jdbi managed object for enabling sql logging:

if (config.isEnableSqlLogging()) {
SqlLogger sqlLogger = new SqlLogger() {
@Override
public void logAfterExecution(StatementContext context) {
log.info("sql {}, parameters {}, timeTaken {} ms", context.getRenderedSql(),
context.getBinding().toString(), context.getElapsedTime(ChronoUnit.MILLIS));
}
};
jdbi.setSqlLogger(sqlLogger);
}

2. Add a config parameter (optional) to control enabling of sql logging as per requirement in various environments:

enableSqlLogging : false




Sunday, 18 October 2020

Puzzle - 100 people standing in a circle - Answer

 If the total number of people is 2^n, the winner will always be the person at the starting point.

You can think of this 2^n behavior with example of 2, 4, 8 etc as a sample.

For the case of 100, the highest 2^n number inside 100 is 64, so we need to eliminate 36 and the next starting person will be the winner.

For eliminating 36 numbers, the last number being removed will be 72 and next starting point will be 73.
So, the answer is 73.


A java program to solve this puzzle is as follows:
@Test
public void testPuzzle () {
int numOfPeople = 8;
int powOfTwo = 1;
while (powOfTwo < numOfPeople) {
powOfTwo = powOfTwo * 2;
}

int winnerIndex = 1;
if (powOfTwo > numOfPeople) {
powOfTwo = powOfTwo/2;
int diff = numOfPeople - powOfTwo;
winnerIndex = diff * 2 + 1;
}
System.out.println(winnerIndex);
}

Friday, 16 September 2016

Java 5 - ReadWriteLock for better performance - ReentrantReadWriteLock and StampedLock


ReadWriteLock was introduced in Java 5 in the form of ReentrantReadWriteLock. A readWriteLock improves performance significantly in a multi-threaded environment when a resource is read more often than it is written.

The concept is simple, multiple threads can do read simultaneously and does not need to be blocked on each other, but if a write operation is going on all the subsequent write/read locks need to wait. A write lock will also wait for already acquired read locks before it gains the exclusive lock on the resource.

A typical use case of ReadWriteLock comes while maintaining a cache that needs to be updated when a new value gets added to the system.

An example cache managed by ReadWriteLock is as follows:


package com.prasune.test.concurrent;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class CachedDataManager {

    private static Map<String, Object> cachedData = new HashMap<>();

    /**
     * Read/write lock to manage cache read/write operations.
     */
    private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    /**
     * Singleton instance
     */
    private static CachedDataManager instance = new CachedDataManager();

    /**
     * private constructor to make class singleton
     */
    private CachedDataManager() {

    }
   
    public CachedDataManager getInstance() {
        return instance;
    }

    /**
     * Fetch cache data using key
     * @param key
     * @return
     */
    public Object getCachedData(String key) {
        Object data = null;
        try {
            readWriteLock.readLock().lock();
            data = cachedData.get(key);           
        } finally {
            readWriteLock.readLock().unlock();
        }
        return data;
    }
   
    /**
     * Update cache  by adding new entry
     * @param key
     * @param data
     */
    public void addToCache(String key, Object data) {
        try {
            readWriteLock.writeLock().lock();
            cachedData.put(key, data);           
        } finally {
            readWriteLock.writeLock().unlock();
        }
    }
}




ReentrantReadWriteLock performs better than the traditional lock mechanism, but it is not fast enough and is too slow at times. So, Java 8 introduced a new ReadWriteLock in the form of StampedLock which uses new set of algorithms and memory management introduced in Java 8.

StampedLock uses a stamp to identify the lock which helps in better performance in managing the read/write lock.

Let us modify our program to use StampedLock:


package com.prasune.test.concurrent;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.StampedLock;

public class CachedDataManager {

    private static Map<String, Object> cachedData = new HashMap<>();

    /**
     * Read/write lock to manage cache read/write operations.
     */
    private final StampedLock readWriteLock = new StampedLock();

    /**
     * Singleton instance
     */
    private static CachedDataManager instance = new CachedDataManager();

    /**
     * private constructor to make class singleton
     */
    private CachedDataManager() {

    }
   
    public CachedDataManager getInstance() {
        return instance;
    }

    /**
     * Fetch cache data using key
     * @param key
     * @return
     */
    public Object getCachedData(String key) {
        Object data = null;
        Long stamp = 0L;
        try {
            stamp = readWriteLock.readLock();
            data = cachedData.get(key);           
        } finally {
            readWriteLock.unlockRead(stamp);
        }
        return data;
    }
   
    /**
     * Update cache  by adding new entry
     * @param key
     * @param data
     */
    public void addToCache(String key, Object data) {
        Long stamp = 0L;
        try {
            stamp = readWriteLock.writeLock();
            cachedData.put(key, data);           
        } finally {
            readWriteLock.unlockWrite(stamp);
        }
    }
}





Java 8 - Producer/Consumer threads using executor framework


Creating a thread consumes a significant amount of memory. In an application where there are lot of client programs, creating a thread per client will not scale. So, Java 5 came up with an executor framework to provide a thread pool for execution limiting the number of threads serving client request at any point of time. This helps in performance and in reducing the memory requirement.

Java 5 also provides blocking queue implementations and we no longer requires to control producer/consumer applications using wait/notify. This is automatically taken care by BlockingQueue implementations.

An example producer/consumer making use of a blocking queue implementation and executor framework is as follows:


package com.prasune.coding.thread;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;

public class TestProducerConsumer {

    private static final int NUM_OF_MSGS = 20;
    private static final BlockingQueue<String> queue 
                                              = new ArrayBlockingQueue<String>(5);
    private static ExecutorService producerPool = Executors.newFixedThreadPool(3);
    private static ExecutorService consumerPool = Executors.newFixedThreadPool(1);

    private static Logger logger =                                                                               Logger.getLogger(TestProducerConsumer.class.getName());

    public static void main(String[] args) {
        Runnable producerTask = () -> {
            try {
                queue.put("test Message");
                System.out.println(Thread.currentThread().getName() 
                                   + " put message queue.size() " + queue.size());
            } catch (InterruptedException e) {
                logger.log(Level.SEVERE, e.getMessage(), e);
            }
        };
        Runnable consumerTask = () -> {
            try {
                System.out.println(Thread.currentThread().getName() 
                                   + " received msg " + queue.take());

            } catch (InterruptedException e) {
                logger.log(Level.SEVERE, e.getMessage(), e);
            }
        };
        try {
            for (int i = 0; i < NUM_OF_MSGS; i++) {
                producerPool.submit(producerTask);
            }
            for (int i = 0; i < NUM_OF_MSGS; i++) {
                consumerPool.submit(consumerTask);
            }
        } finally {
            if (producerPool != null) {
                producerPool.shutdown();
            }
            if (consumerPool != null) {
                consumerPool.shutdown();
            }
        }
    }
}



Tuesday, 9 August 2016

Using Java API for publish/consume Kafka messages


For accessing Java APIs for accessing Kafka, you need to include jar files under kafka<version>/libs in your class path.

A sample message producer java program for Kafka is as follows:


package com.prasune.test.kafka;

import java.util.Date;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 *
 * @author prajohn
 */
public class TestKafkaProducer {
   
    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("zookeeper.connect", "localhost:2181");
        props.put("metadata.broker.list", "localhost:9092");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);

        Producer<String, String> producer = new Producer<String, String>(config);

        String topicName = "test";
        String message = "Test message from Java Producer" + new Date();
        KeyedMessage<String, String> data = new KeyedMessage<String, String>(topicName, message);
        producer.send(data);
       
        producer.close();
    }
   
}



A sample message consumer from Kafka via java program is as follows:


package com.prasune.test.kafka;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import jersey.repackaged.com.google.common.collect.ImmutableMap;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.Message;
import kafka.message.MessageAndMetadata;

/**
 *
 * @author prajohn
 */
public class TestKafkaConsumer {
   
    public static void main(String[] args) {
       
        // specify some consumer properties
        Properties props = new Properties();
        props.put("zookeeper.connect", "localhost:2181");
        props.put("zookeeper.connectiontimeout.ms", "1000000");
        props.put("group.id", "mygroupid2");
       
        // Create the connection to the cluster
        ConsumerConfig cf = new ConsumerConfig(props) ;
        ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(cf) ;
       
        // create 4 partitions of the stream for topic “test”, to allow 4 threads to consume       
        Map<String, List<KafkaStream<byte[], byte[]>>> topicMessageStreams =
                        consumerConnector.createMessageStreams(ImmutableMap.of("test", 4));
        List<KafkaStream<byte[], byte[]>> streams = topicMessageStreams.get("test");

        // create list of 4 threads to consume from each of the partitions
        ExecutorService executor = Executors.newFixedThreadPool(4);

        // consume the messages in the threads
        for(final KafkaStream<byte[], byte[]> stream: streams) {
          executor.submit(new Runnable() {
            public void run() {
              for(MessageAndMetadata msgAndMetadata: stream) {
                System.out.println(new String((byte[]) msgAndMetadata.message()));
              }              
            }
          });
        }
    }
}