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();             }         }     } }

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()));               }                           }           });         }     } }

How to configure and access Kafka via command line

Configuring Kafka with windows:

The commands for configuring kafka are same for both windows and linux, replace .bat with .sh for accessing in linux environment.

Step 1:
Download Apache Kafka binaries and unzip it under say D:/kafka<version>

Step 2:

Go to command prompt and run the kafka commands from windows folder under kafka<version>:

cd kafka<version>/bin/windows

Run Zookeper server:

zookeeper-server-start.bat ../../config/zookeeper.properties

Run Kafka server:

kafka-server-start.bat ../../config/server.properties

Accessing Kafka via command line utilities:

The commands for accessing kafka are same for both windows and linux, replace .bat with .sh for accessing in linux environment.

Create a topic:

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

List the topics created:

kafka-topics.bat --list --zookeeper localhost:2181

Send some message via producer program:

kafka-console-producer.bat --broker-list localhost:9092 --topic test

start a consumer program in a different cmd to see the messages sent from producer:

kafka-console-consumer.bat --zookeeper localhost:2181 --topic test --from-beginning

For newer versions of kafka, below command needs to be used by default for console-consumer program:

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning


To publish/subscribe Kafka messages via Java API, please refer - Using Java API for publish/consume Kafka messages