스프링부트 Kafka 연습 1 - String 주고받기

  spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
      group-id: treesick
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      bootstrap-servers: localhost:9092
      key-serializer: org.apache.kafka.common.serialization.StringDeserializer
      value-serializer: org.apache.kafka.common.serialization.StringDeserializer

consumer
- bootstrap-servers : 카프카 서버의 IP:포트번호
- group-id : 컨슈머그룹 이름 만들어주기. 원하는거 아무거나!
- auto-offset-reset : 컨슈머 그룹에서 소비할 offset 정보가 없을 때, 어떻게 offset을 리셋할 것인지에 대한 설정.
  - latest : 가장 최근 메시지로
  - earliest : 가장 오래된 메시지로
  - none : offset 정보가 없으면 Exception 발생시
- key, value-deserializer : 카프카에서 메세지를 받아올 때, 역직렬화 설정. String이면 StringDeserializer, JSON형식을 받아오면 JsonDeserializer.
  kafkaTemplate의 key-value를 말함.
producer
- bootstrap-servers : 상동
- key,value-deserializer : 반대로 카프카에 데이터를 보낼 때 하는 직렬화 설정.

KafkaProducerService

import lombok.RequiredArgsConstructor;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class KafakaProducerService {
    private static final String TOPIC = "treesick";
    private final KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String message) {
        System.out.println("produce message = " + message);
        kafkaTemplate.send(TOPIC, message);
    }
}

프로듀서 서비스클래스
kafkaTemplate.send를 통해 kafka 서버로 토픽과 메시지를 보냄.

KafkaConsumerService

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

import java.io.IOException;

@Service
public class KafkaConsumerService {

    @KafkaListener(topics = "treesick", groupId = "namu")
    public void consume(String message) throws IOException {
        System.out.println("consume message = " + message);
    }

}

Consumer 서비스클래스
@KafkaListener 어노테이션을 달아서 토픽과 컨슈머그룹아이디 정보를 써줌
메시지가 도달하면 출력을 하도록 코딩

KafkaController

import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import com.sample.kafka.service.KafkaProducerService;

@RestController
@RequiredArgsConstructor
public class KafkaController {
    private final KafkaProducerService producerService;

    @PostMapping("/kafka")
    public String sendMessage(@RequestParam("message") String message) {
        producerService.sendMessage(message);

        return "success";
    }
}

kafka에 메시지를 전달할 수 있도록 restApi를 만들어 주기