kafka docker에 간단 설치 후 Spring boot 연동 테스트

간단하게 Kafka 설치

docker-compose.yml 생성 후 docker-compose up -d를 통해 설치

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_HOST_NAME: wedul.pos
      KAFKA_CREATE_TOPICS: "test:1:1"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

설치된 프로세스 확인

생성된 토픽 확인

- test 토픽이 파티션 1와 replication 1로 생성되었는지 여부 확인

$ docker exec -it simple_kafka_1 bash
$ cd /opt/kafka/bin
$ kafka-topics.sh --describe --topic test --zookeeper simple_zookeeper_1

Spring Boot 프로젝트 생성

카프카 설정

package com.kafka.study.configuration;
 
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.core.env.Environment;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * 카프카 설정
 *
 * @author wedul
 * @since 2019-01-24
 **/
@Configuration
@EnableKafka
@PropertySource("classpath:kafka.properties")
public class KafkaConfiguration {
 
  @Autowired
  private Environment env;
 
  private Map<String, Object> producerConfig() {
    Map<String, Object> config = new HashMap<>();
 
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("bootstrap.servers"));
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
 
    return config;
  }
 
  @Bean
  public KafkaTemplate<String, String > kafkaTemplate() {
    return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(producerConfig()));
  }
 
}
 

카프카 컨트롤러

package com.kafka.study.ctrl;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
 
/**
 * study
 *
 * @author wedul
 * @since 2019-01-24
 **/
@RestController
@Slf4j
public class KafkaCtrl {
 
  private final KafkaTemplate kafkaTemplate;
 
  public KafkaCtrl(KafkaTemplate kafkaTemplate) {
    this.kafkaTemplate = kafkaTemplate;
  }
 
  @PostMapping("/send")
  public ResponseEntity<String> sendMessage(String message) {
    if(!StringUtils.isEmpty(message)) kafkaTemplate.send("test", "Message is " + message);
    log.info(message);
    return ResponseEntity.ok("");
  }
}
 

Console-consumer 모니터링 모드

카프카에 메시지를 send 했을 때 모니터링하기 위한 모드 스크립트 실행

bash-4.4# kafka-console-consumer.sh --bootstrap-server wedul.pos:9092 --topic test

실행해보면 콘솔에 메시지가 전송된것을 확인할 수 있다.

Kafka Licenser 설정 

package com.kafka.study.configuration;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
 
/**
 * study
 *
 * @author wedul
 * @since 2019-01-25
 **/
@Slf4j
@Component
public class ReceiveConfiguration {
 
  @KafkaListener(topics = "test", groupId = "console-consumer-1970")
  public void receive(String payload) {
    log.info("received payload='{}'", payload);
  }
 
}
 

보내면 바로 consumer에서 메시지를 받을 수 있도록 리스너를 설정해보자.

그리고 테스트!

2019-01-25 00:09:43.033  INFO 1760 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 2.0.1
2019-01-25 00:09:43.034  INFO 1760 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : fa14705e51bd2ce5
2019-01-25 00:09:43.041  INFO 1760 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : Cluster ID: 8gNzLx__SHq-4p0b_WsydA
2019-01-25 00:09:43.047  INFO 1760 --- [nio-8080-exec-1] com.kafka.study.ctrl.KafkaCtrl           : babo
2019-01-25 00:09:43.069  INFO 1760 --- [ntainer#0-0-C-1] c.k.s.c.ReceiveConfiguration             : received payload='Message is babo'

git 저장소 : https://github.com/weduls/kafka_example