[실습] 사용자 서비스에서 Kafka로부터 ‘게시글 작성 완료 이벤트’ 구독하기

JSCODE 박재성

2025-12-06

✅ 사용자 서비스에서 Kafka로부터 ‘게시글 작성 완료 이벤트’ 구독하기

1. 사용자 서비스에 Kafka 의존성 추가하기
build.gradle

...

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
	implementation 'org.springframework.boot:spring-boot-starter-web'
	implementation 'org.springframework.kafka:spring-kafka'
	developmentOnly 'org.springframework.boot:spring-boot-devtools'
	runtimeOnly 'com.mysql:mysql-connector-j'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

...



 

2. application.yml에 Kafka 연결을 위한 정보 작성하기
application.yml

server:
  port: 8080

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/user-db 
    username: root 
    password: password 
    driver-class-name: com.mysql.cj.jdbc.Driver
  jpa:
    hibernate:
      ddl-auto: update
    show-sql: true
  kafka:
    # Kafka 서버 주소
    bootstrap-servers: localhost:9092
    # 사용자 서비스에서는 메시지를 consume 하기만 함
    consumer:
      # 메시지의 key 역직렬화 방식 : Kafka에서 받아온 메시지를 String으로 변환
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 메시지의 value 역직렬화 방식 : Kafka에서 받아온 메시지를 String으로 변환
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      
client:
  point-service:
    url: http://localhost:8082

 

3. Kafka로부터 전달받을 메시지 객체 만들기
event/BoardCreatedEvent

public class BoardCreatedEvent {
  private Long userId;

  // 역직렬화(String 형태의 카프카 메시지 -> Java 객체)시 빈생성자 필요함
  public BoardCreatedEvent() {
  }

  // Json 값을 BoardCreatedEvent로 역직렬화하는 메서드
  public static BoardCreatedEvent fromJson(String json) {
    try {
      ObjectMapper objectMapper = new ObjectMapper();
      return objectMapper.readValue(json, BoardCreatedEvent.class);
    } catch (JsonProcessingException e) {
      throw new RuntimeException("JSON 파싱 실패");
    }
  }

  public Long getUserId() {
    return userId;
  }
}

 

4. Kafka에 있는 메시지 Consume 하기
consumer/BoardCreatedEventConsumer

@Component
public class BoardCreatedEventConsumer {
  private final UserService userService;

  public BoardCreatedEventConsumer(UserService userService) {
    this.userService = userService;
  }

  @KafkaListener(
      topics = "board.created",
      groupId = "user-service"
  )
  public void consume(String message) {
    BoardCreatedEvent boardCreatedEvent
        = BoardCreatedEvent.fromJson(message);

    // 게시글 작성 시 활동 점수 10점 추가
    AddActivityScoreRequestDto addActivityScoreRequestDto
        = new AddActivityScoreRequestDto(
            boardCreatedEvent.getUserId(),
            10
        );
    userService.addActivityScore(addActivityScoreRequestDto);
    System.out.println("활동 점수 적립 완료");
  }
}

 
dto/AddActivityScoreRequestDto

public class AddActivityScoreRequestDto {
  private Long userId;
  private int score;

  public AddActivityScoreRequestDto(Long userId, int score) {
    this.userId = userId;
    this.score = score;
  }

  public Long getUserId() {
    return userId;
  }

  public int getScore() {
    return score;
  }
}

 

6. 서버 다시 실행시키기