[Spring Boot] Spring Boot + Kafka 연동

 

공익을 위한 포스팅은 아니고...

계속 눈에 잘근잘근 밟히던 코드를 수정하기 위해 간단하게 카프카를 연동했다

 

수정 이유 

친환경 미션 앱에서 기존에는 미션 완료 api가 호출되면

• 미션 조회 및 완료 가능 여부 확인
• 이미 완료한 미션인지 중복 검사
• UserMission 저장
• 나무 성장 상태 갱신 및 저장
• 배지 획득 여부 확인 

이 모든 게 순서대로(동기적으로) 이루어지고 있었는데

문제는 사용자가 미션 하나를 완료할 때 나무 성장, 배지 확인 같은 후속 작업까지 순차적으로 처리하면
작업이 늘어날수록 응답 시간이 길어지고 장애가 생기면 전체 흐름에 영향을 줄 수 있다는 점이었다.

개발 당시에는 처음 진행한 프로젝트이기도 했고 그냥 순서대로 잘.. 실행되기만 바라면서 코드를 짰는데

돌아보니 고칠 부분이 많다. 

 

그래서 전부 다 비동기 처리하는 건 아니고

UserMission 저장까지는 그대로, 

나무 상태 갱신 및 배지 확인 같은 후속적인 작업을 비동기 처리하기로 결정했다. 

 

 

그래서 kafka가 missionCompletedEvent 를 발행해주면

그걸 들은 Consumer들이 알아서 각각 일을 실행해주면 된다... 

 

 

---

 

Docker Compose 환경설정

우선 환경설정을 위해 docker compose를 다운로드해준다. 

기존 docker는 단일 컨테이너를 관리하는 도구인 반면 얘는 여러 컨테이너를 관리할 수 있다. 

https://docs.docker.com/compose/

Docker Compose

 

 

그리고 내 프로젝트 루트 위치에 docker-compose.yml 파일을 생성해 준다. 

 

docker-compose.yml

services:
  kafka:
    image: apache/kafka:3.9.1
    container_name: ecostep-kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
  kafka-ui:
   image: provectuslabs/kafka-ui:latest
   ports:
     - "8081:8080"
   environment:
     KAFKA_CLUSTERS_0_NAME: local
     KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
   depends_on:
     - kafka

우선 카프카 실행이 잘 되는지 보기 위해 최소 구성인 단일 노드로 작성했다. 

그리고 kafka의 ui를 내 8081포트랑 연결해서, 

이렇게 편하게 내 브로커, 토픽 등을 모니터링할 수 있다. 

 

 

 

이제 터미널을 켜서... 

docker compose up 명령어를 통해 도커를 실행해 준다. 

(반대로 docker compose down은 실행 중인 컨테이너를 멈추고 컨테이너, 이미지 등을 제거한다.)

 

(docker compose up -d 에서 -d는 detached로, 백그라운드 실행 명령어이다. 이걸 입력하면 새로 창을 켤 필요 없이 바로 다음 커맨드를 입력할 수 있다. )

 

그리고 docker compose ps 로 실행 상태 확인, Status가 Up이면 잘 실행되고 있는 것이다. 

 

도커 컴포즈에서도 이렇게 확인 가능..

 

 

 

Spring Boot 설정

그리고 스프링부트로 넘어와서 환경설정을 마무리해준다. 

 

application.yml, 

  kafka:
    bootstrap-servers: localhost:9092

    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

    consumer:
      group-id: ecostep-consumer
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        "[spring.json.trusted.packages]": <이벤트 클래스 패키지 위치>

 

build.gradle

에도 kafka 설정을 추가해주고 gradle 동기화 . . . 

implementation 'org.springframework.kafka:spring-kafka'

 

 

 

이벤트 설계

이제 이벤트를 설계해 볼 건데... 

 

MissionCompletedEvent (이벤트 객체)

public record MissionCompletedEvent(
        Long userId,
        Long missionId,
        LocalDateTime completedAt
) {
}

우선 내 서비스에서의 미션은

누가, 어떤 미션을, 언제 완료했는지 이 세 가지 정보가 필수적이기 때문에 이벤트 객체는 이렇게 만들었다. 

여기서 LocalDateTime이 있기 때문에 JSON으로 직렬화, 역직렬화가 가능하도록 application.yml의 producer, consumer 설정을 JsonSerializer로 해 주었다. 

 

 

Producer 구현

@Component
@RequiredArgsConstructor
public class MissionEventProducer {
    private final KafkaTemplate<String, MissionCompletedEvent>
            kafkaTemplate;

    public void publish(MissionCompletedEvent event) {
        kafkaTemplate.send(
                "mission.completed",
                event.userId().toString(),
                event
        );
    }
}

Spring Kafka에서는 KafkaTemplate을 이용해 producer을 구현할 수 있다. 이 템플릿이 메시지를 보내는 도구이다. 

"mission.completed"는 메시지가 전달될 토픽 이름이며

MissionCompletedEvent 객체에서 userId를 가져오고 객체도 같이 전달한다. 

 

 

Consumer 구현

그리고 consumer는 이 메시지를 받고 후속 작업을 진행할 것들이 대상이 되어야 되는데 

나무 상태 갱신 & 배지 확인

이 있다.

근데 바꿀 코드가 많아서 일단 메시지를 잘 받는지 테스트부터 해보기로... 

@Slf4j
@Component
@RequiredArgsConstructor
public class MissionCompletedConsumer {
    @KafkaListener(
            topics = "mission.completed",
            groupId = "ecostep-consumer"
    )
    public void consume(MissionCompletedEvent event) {
        log.info(
                "미션 완료 이벤트 수신 - userId: {}, missionId: {}, completedAt: {}",
                event.userId(),
                event.missionId(),
                event.completedAt()
        );
    }
}

 

보면 메시지를 보낼 때는 KafkaTemplate, 받을 때는 @KafkaListener를 쓴다.

앞선 application.yml의 spring.kafka 설정으로 KafkaTemplate과 Listener를 자동으로 빈으로 등록하기 때문에, 더 간단하게 작업할 수 있다. 

 

 

Topic 설정 

그리고 configuration을 작성하였는데

@Configuration
public class KafkaTopicConfig {
    @Bean
    public NewTopic missionCompleted(){
        return TopicBuilder.name("mission.completed")
                .partitions(1)
                .replicas(1)
                .build();
    }
}

NewTopic을 사용하면 Spring Kafka의 관리 기능으로 토픽 생성을 요청할 수 있다.

 

Controller 설정

private final MissionEventProducer producer;

    @PostMapping("/mission-completed")
    public Map<String, String> publishTestEvent() {
        MissionCompletedEvent event = new MissionCompletedEvent(
                1L,
                100L,
                LocalDateTime.now()
        );

        producer.publish(event);

        return Map.of("message", "Kafka 이벤트 발행 요청 완료");
    }

해당 api로 요청이 들어오면 테스트용 사용자 ID와 미션 ID로 임의의 이벤트를 발행...

 

 

그리고 서버를 켜주면

이런 로그들이 뜬다. 

 

• Cluster ID: [값]:  Kafka 클러스터를 식별하기 위한 값으로, kafka에 잘 연결되었다
• Successfully joined group: consumer가 consumer group에 참여함
• partitions assigned: consumer가 읽을 파티션을 할당받음
• found no committed offset: 해당 consumer Group에 저장된 offset이 없음

 

 

이후 포스트맨에서 요청을 보냈더니 

요렇게~ 응답을 받을 수 있었다. 

 

 

남은 건 코드 리팩토링.. 

 

---

 

[Java] Spring Boot Kafka 이해하고 활용하기 -2 : 환경 구성 및 사용 예시