Kafka — Fixed Rate Producer and Consumer

In this example, I will show you how to consume the messages from the kafka Real Time.

kafka-topics --bootstrap-server localhost:9092 --create --topic t-fixedrate --partitions 1 --replication-factor 1
Created topic t-fixedrate.

Producer: FixedRateProducer.java

package com.course.kafka.producer;

import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

@Service
public class FixedRateProducer {
   @Autowired
   private KafkaTemplate<String, String> kafkaTemplate;
   
   private AtomicInteger atomicInteger = new AtomicInteger();
   
   @Scheduled(fixedRate = 1000)
   public void sendMessage() {
      int i = atomicInteger.incrementAndGet();
      System.out.println("i is : "+ i);
      kafkaTemplate.send("t-fixedrate", "Fixed rate "+i);
   }
}

MainApp.java

package com.course.kafka;

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class KafkaProducerApplication implements CommandLineRunner{
   

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

   @Override
   public void run(String... args) throws Exception {
   }
}

FixedRateConsumer.java

package com.course.kafka.consumer;

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

@Service
public class HelloKafkaConsumer {

   @KafkaListener(topics = "t-fixedrate")
   public void consume(String message) {
      System.out.println(message);
   }
}

application.yml — auto-offset-reset: earliest will consume from latest offset.

logging:
  pattern:
    console: "[Kafka Core Consumer] %clr(%d{HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:%5p}) %clr(---){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:%wEx}"

spring:
  kafka:
    consumer:
      group-id: default-spring-consumer