Example Code for Connecting Kafka to Spring Boot

Function

Spring Boot is used to produce and consume Kafka clusters.

Sample Code

Use Spring Boot to implement Kafka production and consumption.

@RestController
public class MessageController {
    private final static Logger LOG = LoggerFactory.getLogger(MessageController.class);
    @Autowired
    private KafkaProperties kafkaProperties;
    @GetMapping("/produce")
    public String produce() {
        Producer producerThread = new Producer();
        producerThread.init(this.kafkaProperties);
        producerThread.start();
        String message = "Start to produce messages";
        LOG.info(message);
        return message;
    }
    @GetMapping("/consume")
    public String consume() {
        Consumer consumerThread = new Consumer();
        consumerThread.init(this.kafkaProperties);
        consumerThread.start();
        LOG.info("Start to consume messages");
        // Wait for 180s and close the consumer. Modification can be made during actual execution.
        try {
            Thread.sleep(consumerThread.getThreadAliveTime());
        } catch (InterruptedException e) {
            LOG.info("Occurred InterruptedException: ", e);
        } finally {
            consumerThread.close();
        }
        return String.format("Finished consume messages");
    }
}