[김대호] Sprint11

[kdfasdf]

Spring Event - 파일 업로드 로직 분리하기

• 디스코드잇은 BinaryContent의 메타 데이터(DB)와 바이너리 데이터(FileSystem/S3)를 분리해 저장합니다.
• 만약 지금처럼 두 로직이 하나의 트랜잭션으로 묶인 경우 트랜잭션을 과도하게 오래 점유할 수 있는 문제가 있습니다.
  • 바이너리 데이터 저장 연산은 오래 걸릴 수 있는 연산이며, 해당 연산이 끝날 때까지 트랜잭션이 대기해야합니다.
• 따라서 Spring Event를 활용해 메타 데이터 저장 트랜잭션으로부터 바이너리 데이터 저장 로직을 분리하여, 메타데이터 저장 트랜잭션이 종료되면 바이너리 데이터를 저장하도록 변경합니다.
• • • BinaryContentStorage.put을 직접 호출하는 대신 BinaryContentCreatedEvent를 발행하세요.
  • BinaryContentCreatedEvent를 정의하세요.
    • BinaryContent 메타 정보가 DB에 잘 저장되었다는 사실을 의미하는 이벤트입니다.
  • 다음의 메소드에서 BinaryContentStorage를 호출하는 대신 BinaryContentCreatedEvent를 발행하세요.
    • UserService.create/update
    • MessageService.create
    • BinaryContentService.create
  • ApplicationEventPublisher를 활용하세요.
• • • 이벤트를 받아 실제 바이너리 데이터를 저장하는 리스너를 구현하세요.
  • 이벤트를 발행한 메인 서비스의 트랜잭션이 커밋되었을 때 리스너가 실행되도록 설정하세요.
  • BinaryContentStorage를 통해 바이너리 데이터를 저장하세요.
• • 바이너리 데이터 저장 성공 여부를 알 수 있도록 메타 데이터를 리팩토링하세요.

• BinaryContent에 바이너리 데이터 업로드 상태 속성(status)을 추가하세요.

  - PROCESSING: 업로드 중
     - 기본값입니다.
  - SUCCESS: 업로드 완료
  - FAIL: 업로드 실패

-- schema.sql
CREATE TABLE binary_contents
(
    id           uuid PRIMARY KEY,
    created_at   timestamp with time zone NOT NULL,
    updated_at timestamp with time zone,
    file_name    varchar(255)             NOT NULL,
    size         bigint                   NOT NULL,
    content_type varchar(100)             NOT NULL,
    status       varchar(20)              NOT NULL
);

-- ALTER TABLE binary_contents
--      ADD COLUMN updated_at timestamp with time zone;
-- ALTER TABLE binary_contents
--      ADD COLUMN status varchar(20) NOT NULL;

• BinaryContent의 상태를 업데이트하는 메소드를 정의하세요.

  - 트랜잭션 전파 범위에 유의하세요.

• • • 바이너리 데이터 저장 성공 여부를 메타 데이터에 반영하세요.
  • 성공 시 BinaryContent의 status를 SUCCESS로 업데이트하세요.
  • 실패 시 BinaryContent의 status를 FAIL로 업데이트하세요.

Spring Event - 알림 기능 추가하기

• 1)채널에 새로운 메시지가 등록되거나 2)권한이 변경된 경우 이벤트를 발행해 알림을 받을 수 있도록 구현합니다.
• • • 채널에 새로운 메시지가 등록된 경우 알림을 받을 수 있도록 리팩토링하세요.
  • MessageCreatedEvent를 정의하고 새로운 메시지가 등록되면 이벤트를 발행하세요.
  • 사용자 별로 관심있는 채널의 알림만 받을 수 있도록 ReadStatus 엔티티에 채널 알림 여부 속성(notificationEnabled)을 추가하세요.

  - PRIVATE 채널은 알림 여부를 true로 초기화합니다.
  - PUBLIC 채널은 알림 여부를 false로 초기화합니다.

-- schema.sql
CREATE TABLE read_statuses
(
    id           uuid PRIMARY KEY,
    created_at   timestamp with time zone NOT NULL,
    updated_at   timestamp with time zone,
    user_id      uuid                     NOT NULL,
    channel_id   uuid                     NOT NULL,
    last_read_at timestamp with time zone NOT NULL,
    notification_enabled boolean NOT NULL,
    UNIQUE (user_id, channel_id)
);

-- ALTER TABLE read_statuses
--      ADD COLUMN notification_enabled boolean NOT NULL;

• 알림 여부를 수정할 수 있게 ReadStatusUpdateRequest를 수정하세요.

  - 알림이 활성화 되어 있는 경우

  - 알림이 활성화 되어 있지 않은 경우

• • 사용자의 권한(Role)이 변경된 경우 알림을 받을 수 있도록 리팩토링하세요.
  • RoleUpdatedEvent를 정의하고 권한이 변경되면 이벤트를 발행하세요.
• • 알림 API를 구현하세요.
  • NotificationDto를 정의하세요.

  - receiverId: 알림을 수신할 User의 id입니다.

• 알림 조회
  • 엔드포인트: GET /api/notifications
  • 요청
    • 헤더: 엑세스 토큰
  • 응답
    • 200 List
    • 401 ErrorResponse
• 알림 확인
  • 엔드포인트: DELETE /api/notifications/{notificationId}
  • 요청
    • 헤더: 엑세스 토큰
  • 응답
    • 204 Void
    • 인증되지 않은 요청: 401 ErrorResponse
    • 인가되지 않은 요청: 403 ErrorResponse
      • 요청자 본인의 알림에 대해서만 수행할 수 있습니다.
    • 알림이 없는 경우: 404 ErrorResponse
• • 알림이 필요한 이벤트가 발행되었을 때 알림을 생성하세요.
  • 이벤트를 처리할 리스너를 구현하세요.

public class NotificationRequiredEventListener {

  @TransactionalEventListener
  public void on(MessageCreatedEvent event) {...}

  @TransactionalEventListener
  public void on(RoleUpdatedEvent event) {...}
}

  - on(MessageCreatedEvent)
     - 해당 채널의 알림 여부를 활성화한 ReadStatus를 조회합니다.

     - 해당 ReadStatus의 사용자들에게 알림을 생성합니다.

# 알림 예시
title: "보낸 사람 (#채널명)"
content: "메시지 내용"

     - 단, 해당 메시지를 보낸 사람은 알림 대상에서 제외합니다.

  - on(RoleUpdatedEvent)
     - 권한이 변경된 당사자에게 알림을 생성합니다.

# 알림 예시
title: "권한이 변경되었습니다."
content: "USER -> CHANNEL_MANAGER"

비동기 적용하기

• • 비동기를 적용하기 위한 설정(AsyncConfig) 클래스를 구현하세요.
  • @EanbleAsync 어노테이션을 활용하세요.
  • TaskExecutor를 Bean으로 등록하세요.
  • TaskDecorator를 활용해 MDC의 Request ID, SecurityContext의 인정 정보가 비동기 스레드에서도 유지되도록 구현하세요.
• • 앞서 구현한 Event Listener를 비동기적으로 처리하세요.
  • [@async](https://github.com/async) 어노테이션을 활용하세요.
• • 동기 처리와 비동기 처리 간 성능 차이를 비교해보세요.
  • 파일 업로드 로직에 의도적인 지연(Thread.sleep(…))을 발생시키세요.

// LocalBinaryContentStorage
public UUID put(UUID binaryContentId, byte[] bytes) {
    try {
      Thread.sleep(3000);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new RuntimeException("Thread interrupted while simulating delay", e);
    }
    ...
}

• 메시지 생성 API의 실행 시간을 측정해보세요.
  • [@timed](https://github.com/timed) 어노테이션을 메소드에 추가합니다.

// MessageController
  @Timed("message.create.async")
  @PostMapping(consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
  public ResponseEntity<MessageDto> create(...) {...}

  - Actuator 설정을 추가합니다.

# application.yaml
management:
    ...
  observations:
    annotations:
      enabled: true

     - /actuator/metrics/message.create.async 에서 측정된 시간을 확인할 수 있습니다.

• [@EnableAsync](https://github.com/EnableAsync)를 활성화 / 비활성화 해보면서 동기 / 비동기 처리 간 응답 속도의 차이를 확인해보세요.

비동기 실패 처리하기

• 비동기로 처리하는 로직이 실패하는 경우 사용자에게 즉각적인 에러 전파가 되지 않을 가능성이 높습니다.
• 따라서 비동기로 처리하는 로직은 자동 재시도 전략을 통해 더 견고하게 구현해야 합니다.
• 또, 실패하더라도 그 사실을 명확하게 기록해두어야 에러에 대응할 수 있습니다.
• • S3를 활용해 바이너리 데이터 저장 시 자동 재시도 매커니즘을 구축하세요.
  • Spring Retry를 위한 환경을 구성하세요.
    • org.springframework.retry:spring-retry 의존성을 추가하세요.
    • @EnableRetry 어노테이션을 활용해 Spring Retry를 활성화 하세요.
  • 바이너리 데이터를 저장하는 메소드에 [@retryable](https://github.com/retryable) 어노테이션을 사용해 재시도 정책(횟수, 대기 시간 등)을 설정하세요.
• • • 재시도가 모두 실패했을 때 대응 전략을 구축하세요.
  • [@recover](https://github.com/recover) 어노테이션을 활용하세요.
  • 실패 정보를 관리자에게 통지하세요.

캐시 적용하기

• • Caffeine 캐시를 위한 환경을 구성하세요.
  • org.springframework.boot:spring-boot-starter-cache 의존성을 추가하세요.
  • com.github.ben-manes.caffeine:caffeine 의존성을 추가하세요.
  • application.yaml 설정 또는 Bean을 통해 캐시 Caffeine 캐시를 설정하세요.
• • [@Cacheable](https://github.com/Cacheable) 어노테이션을 활용해 캐시가 필요한 메소드에 적용하세요.
  • 사용자별 채널 목록 조회
  • 사용자별 알림 목록 조회
  • 사용자 목록 조회
• • 데이터 변경 시, 캐시를 갱신 또는 무효화하는 로직을 구현하세요.
  • [@CacheEvict](https://github.com/CacheEvict), [**[@cACHEpUT](https://github.com/cACHEpUT)**](https://github.com/cACHEpUT), CacheManager 등을 활용하세요.
  • 예시:
    • 새로운 채널 추가/수정/삭제 → 채널 목록 캐시 무효화
    • 알림 추가/삭제 → 알림 목록 캐시 무효화
    • 사용자 추가/로그인/로그아웃 → 사용자 목록 캐시 무효화
• • 캐시 적용 전후의 차이를 비교해보세요.
  • 로그를 통해 SQL 실행 여부를 확인해보세요.
• • Spring Actuator를 활용해 캐시 관련 통계 지표를 확인해보세요.
  • Caffein Spec에 recordStats 옵션을 추가하세요.

  # application.yaml
  cache:
        ...
    caffeine:
      spec: >
        maximumSize=100,
        expireAfterAccess=600s,
        recordStats

• /actuator/caches, /actuator/metrics/cache.* 를 통해 캐시 관련 데이터를 확인해보세요.

심화

유의사항

• 이번 실습은 분산 환경을 대비한 기술 도입을 목표로 합니다. 특히, 여러 서버나 인스턴스로 구성되는 시스템에서 발생할 수 있는 문제를 해결하는 데 필요한 기술을 학습합니다.
  • Kafka를 통한 이벤트 발행/구독
  • Redis를 활용한 전역 캐시 저장소 구성
• 이번 미션에서는 Kafka, Redis의 세부 설정이나 고급 기능보다는, 기술을 간단하게 적용하고, 분산 환경의 필요성과 기존 시스템의 한계를 이해하는 데 집중해주세요.

Spring Kafka 도입하기

• 회원이 늘어나면서 알림 연산량이 급증해 알림 기능만 별도의 마이크로 서비스로 분리하기로 결정했다고 가정해봅시다.
• 이제 알림 서비스와 메인 서비스는 완전히 분리된 서버이므로 Spring Event만을 통해서 이벤트를 발행/소비할 수 없습니다.
• 따라서 메인 서비스에서 Kafka를 통해 서버 외부로 이벤트를 발행하고, 알림 서비스에서는 서버 외부의 이벤트를 소비할 수 있도록 해야합니다.
• • Kafka 환경을 구성하세요.
  • Docker Compose를 활용해 Kafka를 구동하세요.

# docker-compose-kafka.yaml
# https://developer.confluent.io/confluent-tutorials/kafka-on-docker/#the-docker-compose-file
services:
  broker:
    image: apache/kafka:latest
    hostname: broker
    container_name: broker
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
      KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

docker compose -f docker-compose-kafka.yaml up -d

• Spring Kafka 의존성을 추가하고, application.yml에 Kafka 설정을 추가하세요.

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

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

• • Spring Event를 Kafka로 발행하는 리스너를 구현하세요.
  • NotificationRequiredEventListener는 비활성화하세요.
  • KafkaProduceRequiredEventListener를 구현하세요.

package com.sprint.mission.discodeit.event.kafka;

@Slf4j
@RequiredArgsConstructor
@Component
public class KafkaProduceRequiredEventListener {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;

  @Async("eventTaskExecutor")
  @TransactionalEventListener
  public void on(MessageCreatedEvent event) {
    ...
    String payload = objectMapper.writeValueAsString(event);
    kafkaTemplate.send("discodeit.MessageCreatedEvent", payload);
    ...
  }

  @Async("eventTaskExecutor")
  @TransactionalEventListener
  public void on(RoleUpdatedEvent event) {...}

  @Async("eventTaskExecutor")
  @EventListener
  public void on(S3UploadFailedEvent event) {...}
}

  - Spring Event를 Kafka 메시지로 변환해 전송하는 중계 구조입니다.

60f20el76-image.png

• • Kafka Console을 통해 Kafka 이벤트가 잘 발행되는지 확인해보세요.
  1. broker 컨테이너 쉘 접속

docker exec -it -w /opt/kafka/bin broker sh

  2. 토픽 리스트 확인 (실행위치: /opt/kafka/bin)

./kafka-topics.sh --list --bootstrap-server broker:29092

__consumer_offsets
discodeit.MessageCreatedEvent
...

  3. 특정 토픽 이벤트 구독 및 대기 (실행위치: /opt/kafka/bin)

./kafka-console-consumer.sh --topic discodeit.MessageCreatedEvent --from-beginning --bootstrap-server broker:29092

...

• • Kafka 토픽을 구독해 알림을 생성하는 리스너를 구현하세요.
  • 이 리스너는 메인 서비스와 별도의 서버로 구성된 알림 서비스라고 가정합니다.
  • NotificationRequiredTopicListener를 구현하세요.

package com.sprint.mission.discodeit.event.kafka;

@Slf4j
@RequiredArgsConstructor
@Component
public class NotificationRequiredTopicListener {
  ...
  private final ObjectMapper objectMapper;

  @KafkaListener(topics = "discodeit.MessageCreatedEvent")
  public void onMessageCreatedEvent(String kafkaEvent) {
    try {
      MessageCreatedEvent event = objectMapper.readValue(kafkaEvent,
          MessageCreatedEvent.class);
            ...
    } catch (JsonProcessingException e) {
      throw new RuntimeException(e);
    }
  }

  @KafkaListener(topics = "discodeit.RoleUpdatedEvent")
  public void onRoleUpdatedEvent(String kafkaEvent) {...}

  @KafkaListener(topics = "discodeit.S3UploadFailedEvent")
  public void onS3UploadFailedEvent(String kafkaEvent) {...}
}

  - 기존 @EventListener 기반 로직을 제거하고 @KafkaListener로 대체하세요.

Redis Cache 도입하기

• 대용량 트래픽을 감당하기 위해 서버의 인스턴스를 여러 개로 늘렸다고 가정해봅시다.

• Caffeine과 같은 로컬 캐시는 서로 다른 서버에서 더 이상 활용할 수 없습니다.

  따라서 Redis를 통해 전역 캐시 저장소를 구성합니다.

• • • Redis 환경을 구성하세요.
  • Docker Compose를 활용해 Redis를 구동하세요.

# docker-compose-redis.yml
# https://developer.confluent.io/confluent-tutorials/kafka-on-docker/#the-docker-compose-file
services:
  redis:
    image: redis:7.2-alpine
    container_name: redis
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    command: redis-server --appendonly yes

volumes:
  redis-data:

docker compose -f docker-compose-redis.yml up -d

• Redis 의존성을 추가하고, application.yml에 Redis 설정을 추가하세요.

implementation 'com.github.ben-manes.caffeine:caffeine'
implementation 'org.springframework.boot:spring-boot-starter-data-redis'

# application.yaml
spring:
  ...
  cache:
    type: caffeine
    type: redis
    cache-names:
      - channels
      - notifications
      - users
    caffeine:
      spec: >
        maximumSize=100,
        expireAfterAccess=600s,
        recordStats
    redis:
      enable-statistics: true
  data:
    redis:
      host: ${REDIS_HOST:localhost}
      port: ${REDIS_PORT:6379}

• 직렬화 설정을 위해 다음과 같이 Bean을 선언하세요.

// CacheConfig
@Bean
public RedisCacheConfiguration redisCacheConfiguration(ObjectMapper objectMapper) {
  ObjectMapper redisObjectMapper = objectMapper.copy();
  redisObjectMapper.activateDefaultTyping(
      LaissezFaireSubTypeValidator.instance,
      DefaultTyping.EVERYTHING,
      As.PROPERTY
  );

  return RedisCacheConfiguration.defaultCacheConfig()
      .serializeValuesWith(
          RedisSerializationContext.SerializationPair.fromSerializer(
              new GenericJackson2JsonRedisSerializer(redisObjectMapper)
          )
      )
      .prefixCacheNameWith("discodeit:")
      .entryTtl(Duration.ofSeconds(600))
      .disableCachingNullValues();
}

• • DataGrip을 통해 Redis에 저장된 캐시 정보를 조회해보세요.

주요 변경사항

스크린샷

[image](https://github.com/codeit-bootcamp-spring/4-sprint-mission/pull/%EC%9D%B4%EB%AF%B8%EC%A7%80url)

멘토에게

• 셀프 코드 리뷰를 통해 질문 이어가겠습니다.

[spring-kang]

AppConfig로 빼서 진행하신 부분 너무 잘하셨습니다. 이렇게하면 컨피그 파일에 대한 책임분리가 잘 이루어집니다.

[spring-kang]

대부분의 기능을 잘 구현하셨습니다. 고생하셨습니다.