알림 서비스 구현하기(Spring Batch, SSE, Kafka)

알림 서비스 구현하기(Spring Batch, SSE, Kafka)

미흡한 부분이 있다면 언제든 댓글에 남겨주세요😊 수정보완하겠습니다!
앞으로 진행될 순서는 다음과 같습니다.

1. 왜 이 기능을 구현했나요?
2. 왜 이 기술을 사용했나요?
3. 어떻게 구현했나요?
4. 더 알아보기

1. 왜 이 기능을 구현했나요?

저희 프로젝트 주제는 펀딩 서비스였습니다. 펀딩에 성공하면 펀딩을 마감하고 쿠폰을 생성하여 알림을 보내주는 로직를 작성해야 했습니다. 또한, 관심 펀딩이 마감되기 전, 쿠폰이 만료되기 전에 알림을 보내 펀딩 참여를 유도하고, 쿠폰 사용을 촉진하고자 하였습니다.

2. 왜 이 기술을 사용했나요?

Spring Batch

• 일괄 처리 (Batch Processing):
  • Spring Batch는 대용량의 데이터를 효과적으로 일괄 처리하는 데에 특화되어 있습니다. 대량의 데이터를 효율적으로 처리하고, 장애 복구 및 재시도 메커니즘을 통해 안정적인 일괄 처리를 제공합니다.
• 스프링 생태계와 통합:
  • 스프링 Batch는 스프링 프레임워크와 통합되어 있어서 스프링의 핵심 원리와 일관성을 유지하면서 배치 작업을 개발할 수 있습니다. 스프링의 다양한 기능과 함께 사용 가능하며, 스프링의 의존성 주입과 같은 기능을 활용할 수 있습니다.
• 장애 처리와 롤백 지원:
  • Spring Batch는 일괄 처리 작업 중에 발생할 수 있는 장애에 대한 처리와 롤백을 지원합니다. 이를 통해 안정적인 데이터 처리를 보장하고, 장애 시에도 데이터 일관성을 유지할 수 있습니다.
• 병렬 처리 및 확장성:
  • 대용량 데이터 처리를 위해 Spring Batch는 병렬 처리 및 확장성을 제공합니다. 작업을 여러 쓰레드 또는 노드에서 병렬로 처리할 수 있어 성능을 향상시킬 수 있습니다.

SSE

SSE vs. WebSockets 를 비교해보면

• SSE:
  • 용도: 단방향 실시간 이벤트 전송에 적합.
  • 장점: 간단하게 구현 가능. 브라우저의 내장 기능 활용 가능.
  • 한계: 단방향 통신만 가능하며, 연결 상태 유지가 어려움.
• WebSockets:
  • 용도: 양방향 통신이 필요한 실시간 애플리케이션에 적합.
  • 장점: 실시간 양방향 통신 가능. 연결 상태를 유지할 수 있음.
  • 한계: 구현이 복잡하고, 일부 환경에서는 방화벽 문제가 발생할 수 있음.

알림의 경우 서버에서 클라이언트로 단방향으로 이동하기 때문에 SSE가 적합합니다.

Kafka

매칭 서비스 구현하기(RabbitMQ, Unit Test)
채팅 서비스 구현하기(NoSQL, WebSocket, Kafka)
두 경우가 합쳐졌다.

1. 서버 분리(배치 서버에서 백엔드 서버로 이벤트를 보냄)
2. 비동기 처리
3. 실시간 처리
4. 데이터 유실 방지

3. 어떻게 구현했나요?

쿠폰 마감 알림에 대한 예시입니다.

docker-compose.yml

# compose 파일 버전
version: '3'
services:
  # 서비스 명
  zookeeper:
    # 사용할 이미지
    image: wurstmeister/zookeeper
    # 컨테이너명 설정
    container_name: zookeeper
    # 접근 포트 설정 (컨테이너 외부:컨테이너 내부)
    ports:
      - "2181:2181"
  # 서비스 명
  kafka:
    # 사용할 이미지
    image: wurstmeister/kafka
    # 컨테이너명 설정
    container_name: kafka
    # 접근 포트 설정 (컨테이너 외부:컨테이너 내부)
    ports:
      - "9092:9092"
    # 환경 변수 설정
    environment:
      KAFKA_ADVERTISED_HOST_NAME: localhost
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    # 볼륨 설정
    volumes:
      - /var/run/docker.sock
    # 의존 관계 설정
    depends_on:
      - zookeeper

프론트

// SSE 알림
    const [, setEventSource] = useState<EventSource | undefined>(undefined);

    const handleGetAlarm = () => {
        getNotificationExist().then((res) => {
            if (res.data) {
                setIsNotificationExist(true);
            } else {
                setIsNotificationExist(false);
            }
        });
    };

    useEffect(() => {
        if (user.userId !== 0) {
            handleGetAlarm();
            const source = new EventSource(
                `http://localhost:8081/api/notification/subscribe/${user.userId}`,
            );
            setEventSource(source);

            source.addEventListener("open", function (event) {
                console.log("connection opened", event);
            });

            source.addEventListener("sse", function (event) {
                console.log("sse", event.data);
                handleGetAlarm();
            });

            source.addEventListener("error", function (event) {
                console.log("error", event);
                source.close();
            });
        }
    }, []);

서버로부터의 실시간 알림을 받기 위해 SSE를 사용하는 부분입니다. 컴포넌트가 렌더링될 때 초기 알림 상태를 확인하고, 유저에게 알림이 있을 경우 SSE를 통해 실시간으로 알림을 수신합니다.

백엔드 서버

NotificationController

@GetMapping(value = "/subscribe/{userId}")
public SseEmitter subscribe(@PathVariable Long userId) {
    log.info("subscribe");
    return sseService.connectNotification(userId);
}

클라이언트가 이 엔드포인트로 GET 요청을 보내면, 해당 유저의 SSE (Server-Sent Events) 연결을 설정하고, 이를 나타내는 SseEmitter 객체를 반환합니다. 이 연결을 통해 서버는 해당 유저에게 실시간 알림을 전송할 수 있게 됩니다.

SseService

@Slf4j
@Service
@RequiredArgsConstructor
public class SseService {

    private final static String SSE_NAME = "sse";

    private final EmitterRepository emitterRepository;
    private static final Long DEFAULT_TIMEOUT = 60L * 1000 * 60;

    public void send(Long notificationId, Long receiverId) {
        log.info("SseService - send 시도");
        emitterRepository.get(receiverId).ifPresentOrElse(it -> {
                try {
                    it.send(SseEmitter.event()
                        .id(notificationId.toString())
                        .name(SSE_NAME)
                        .data("send notification"));
                    log.info("SseService - send 성공");
                } catch (IOException exception) {
                    emitterRepository.delete(receiverId);
                    throw new SseConnectException();
                }
            },
            () -> log.info("No emitter founded")
        );
    }

    public SseEmitter connectNotification(Long userId) {
        SseEmitter emitter = new SseEmitter(DEFAULT_TIMEOUT);
        emitterRepository.save(userId, emitter);
        emitter.onCompletion(() -> emitterRepository.delete(userId));
        emitter.onTimeout(() -> emitterRepository.delete(userId));

        try {
            log.info("connectNotification - send");
            emitter.send(SseEmitter.event()
                .id("id")
                .name(SSE_NAME)
                .data("connect completed"));
        } catch (IOException exception) {
            throw new SseConnectException();
        }
        return emitter;
    }

}

• send 메서드:
  • 특정 알림을 특정 수신자에게 보내는 역할을 합니다.
  • 수신자의 SseEmitter를 찾아 데이터를 보내고, 연결이 실패하면 예외를 처리하고 연결을 삭제합니다.
• connectNotification 메서드:
  • 특정 사용자의 SSE 연결을 설정하고 해당 SseEmitter 객체를 반환합니다.
  • 연결 설정 후, 연결 완료 메시지를 보내고, 연결이 종료되거나 타임아웃이 발생하면 연결을 삭제합니다.

KafkaConsumerConfig

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Value("${kafka.server}")
    private String BOOTSTRAP_SERVER;
    private static final String GROUP_ID = "group";

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

}

카프카 설정파일을 작성해줍니다. 백엔드 서버에서 카프카는 배치 서버에서 생성된 알림을 받습니다.

KafkaConsumer

@Slf4j
@Component
@RequiredArgsConstructor
public class KafkaConsumer {

    private static final String TOPIC_NAME = "sse";
    private final ObjectMapper objectMapper = new ObjectMapper();
    private final SseService sseService;

    @KafkaListener(topics = TOPIC_NAME)
    public void listenMessage(String jsonMessage) {
        try {
            SseDto message = objectMapper.readValue(jsonMessage, SseDto.class);
            sseService.send(message.getNotificationId(), message.getUserId());
            log.info(">>> UserId {}, NotificationId {}", message.getUserId(), message.getNotificationId());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Kafka 토픽에서 메시지를 수신하면서, 수신한 메시지를 처리하여 SSE를 통해 클라이언트에게 알림을 전달하는 역할을 합니다.

배치서버

배치 서버에서는 알림을 생성합니다. 쿠폰 만료 알림의 경우에는 매일 아침 9시에 만료 기한이 하루 전인 쿠폰에 대하여 알림을 보내는 역할을 합니다.

BatchConfig

@EnableBatchProcessing
@Configuration
public class BatchConfig {

    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor(JobRegistry jobRegistry) {
        JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor = new JobRegistryBeanPostProcessor();
        jobRegistryBeanPostProcessor.setJobRegistry(jobRegistry);
        return jobRegistryBeanPostProcessor;
    }
}

Spring Batch에서 사용되는 기본 설정을 활성화하고, 배치 작업을 등록 및 관리하기 위한 JobRegistryBeanPostProcessor 빈을 설정합니다.

스케줄링

@Slf4j
@Component
@RequiredArgsConstructor
public class SendNotificationBeforeExpireCouponScheduler {

    private final Job sendNotificationBeforeExpireCouponJob;
    private final JobLauncher jobLauncher;

    // 매일 오전 9시 마다 실행
    @Scheduled(cron = "0 0 9 * * *")
    public void executeJob () {
        try {
            log.info("sendNotificationBeforeExpireCouponJob start");
            jobLauncher.run(
                sendNotificationBeforeExpireCouponJob,
                new JobParametersBuilder()
                    .addString("datetime", LocalDateTime.now().toString())
                    .toJobParameters()  // job parameter 설정
            );
            log.info("successfully complete sendNotificationBeforeExpireCouponJob");
        } catch (JobExecutionException ex) {
            ex.printStackTrace();
        }
    }
}

@Scheduled 어노테이션을 사용하여 매일 오전 9시에 실행되는 스케줄러를 정의한 것입니다. 해당 스케줄러는 배치 작업을 실행합니다.

배치처리

@Slf4j
@RequiredArgsConstructor
@Configuration
public class SendNotificationBeforeExpireCouponJobConfig {
    private final int CHUNK_SIZE = 10;
    private final DateTimeFormatter dateFormat = DateTimeFormatter.ofPattern("yyyy년 MM월 dd일");
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final EntityManagerFactory entityManagerFactory;
    private final SendNotificationItemWriter sendNotificationItemWriter;

    @Bean
    public Job sendNotificationBeforeExpireCouponJob() {
        return this.jobBuilderFactory.get("sendNotificationBeforeExpireCouponJob")
            .start(addExpireCouponNotificationStep())
            .next(sendExpireCouponNotificationStep())
            .build();
    }

    @Bean
    public Step addExpireCouponNotificationStep() {
        return this.stepBuilderFactory.get("addExpireCouponNotificationStep")
            .<Coupon, Notification>chunk(CHUNK_SIZE)
            .reader(addExpireCouponNotificationItemReader())
            .processor(addExpireCouponNotificationItemProcessor())
            .writer(addExpireCouponNotificationItemWriter())
            .build();
    }

    @Bean
    public JpaPagingItemReader<Coupon> addExpireCouponNotificationItemReader() {
        return new JpaPagingItemReaderBuilder<Coupon>()
            .name("addExpireCouponNotificationItemReader")
            .entityManagerFactory(entityManagerFactory)
            .pageSize(CHUNK_SIZE)
            .queryString("select c from Coupon c " +
                "join fetch c.funding f " +
                "join fetch c.user u " +
                "where c.couponStatus = :couponStatus " +
                "and c.couponExpirationDate <= :couponExpirationDate ")
            .parameterValues(
                Map.of("couponStatus", CouponStatus.ACTIVE, "couponExpirationDate", LocalDate.now().plusDays(1L)))
            .build();
    }

    @Bean
    public ItemProcessor<Coupon, Notification> addExpireCouponNotificationItemProcessor() {
        log.info("addExpireCouponNotificationItemProcessor 실행");
        return coupon -> {
            String message = coupon.getFunding().getFundingTitle() + NotificationType.COUPON_DEADLINE.getMessage()
                + dateFormat.format(coupon.getCouponExpirationDate());
            return Notification.create(coupon.getUser(), coupon.getFunding(), message,
                NotificationType.COUPON_DEADLINE);
        };
    }

    @Bean
    public JpaItemWriter<Notification> addExpireCouponNotificationItemWriter() {
        return new JpaItemWriterBuilder<Notification>()
            .entityManagerFactory(entityManagerFactory)
            .build();
    }

    @Bean
    public Step sendExpireCouponNotificationStep() {
        log.info("sendNotificationStep 실행");
        return this.stepBuilderFactory.get("sendExpireCouponNotificationStep")
            .<Notification, Notification>chunk(CHUNK_SIZE)
            .reader(sendExpireCouponNotificationItemReader())
            .writer(sendNotificationItemWriter)
            .taskExecutor(new SimpleAsyncTaskExecutor())
            .build();
    }

    @Bean
    public SynchronizedItemStreamReader<Notification> sendExpireCouponNotificationItemReader() {
        log.info("sendExpireCouponNotificationItemReader 실행");
        JpaCursorItemReader<Notification> itemReader = new JpaCursorItemReaderBuilder<Notification>()
            .name("sendExpireCouponNotificationItemReader")
            .entityManagerFactory(entityManagerFactory)
            .queryString("select n from Notification n "
                + "join fetch n.user u "
                + "where n.notificationType = :notificationType "
                + "and n.notificationSent = :notificationSent")
            .parameterValues(Map.of("notificationType", NotificationType.COUPON_DEADLINE, "notificationSent", false))
            .build();

        return new SynchronizedItemStreamReaderBuilder<Notification>()
            .delegate(itemReader)
            .build();
    }

}

• sendNotificationBeforeExpireCouponJob: 배치 작업을 정의하는 메서드입니다. 두 개의 스텝을 포함하고 있습니다.
• addExpireCouponNotificationStep: 쿠폰 만료 전 알림을 추가하는 스텝을 정의하는 메서드입니다.
• sendExpireCouponNotificationStep: 알림을 전송하는 스텝을 정의하는 메서드입니다.

1. Chunk-oriented Processing:

• Chunk란?
  • Chunk는 일괄 처리에서 한 번에 처리될 레코드의 묶음을 나타냅니다.
  • 대용량 데이터를 처리할 때, 한 번에 하나씩이 아니라 묶음 단위로 처리하면 성능이 향상됩니다.
• Chunk-oriented Processing 특징:
  • Reader, Processor, Writer 구성: ItemReader, ItemProcessor, ItemWriter는 각각 레코드를 읽고 가공하며 쓰는 역할을 수행합니다.
  • Transaction 범위: Chunk는 하나의 트랜잭션 내에서 처리됩니다. 성공한 청크 내의 모든 레코드는 커밋되고, 실패 시 롤백됩니다.
  • 병렬 처리: Chunk 단위로 병렬 처리가 가능하며, taskExecutor를 설정하여 멀티스레드 환경에서 동작할 수 있습니다.

2. Tasklet-oriented Processing:

• Tasklet이란?
  • Tasklet은 단일한 작업을 수행하는 단위로, 개발자가 직접 구현해야 합니다.
  • Tasklet은 하나의 트랜잭션 안에서 단일 작업을 수행하고, 성공 또는 실패에 따라 트랜잭션을 커밋 또는 롤백합니다.
• Tasklet-oriented Processing 특징:
  • 단일 작업: 하나의 Tasklet이 하나의 트랜잭션에서 수행됩니다.
  • 병렬 처리: Chunk와는 달리 Tasklet은 일반적으로 병렬 처리가 어렵습니다. 각 Tasklet이 독립적인 트랜잭션을 갖기 때문입니다.
  • 트랜잭션 처리: Tasklet은 성공 시 커밋, 실패 시 롤백이 일어납니다.

3. SynchronizedItemStreamReader:

• SynchronizedItemStreamReader란?
  • 멀티스레드 환경에서 ItemStreamReader를 사용할 때, 각 스레드 간에 데이터의 충돌을 방지하기 위해 사용됩니다.
  • SynchronizedItemStreamReader는 내부적으로 synchronized 키워드를 사용하여 여러 스레드 간의 순차적인 데이터 읽기를 보장합니다.

4. SimpleAsyncTaskExecutor:

• SimpleAsyncTaskExecutor란?
  • 스프링 프레임워크에서 제공하는 간단한 비동기 태스크 실행자입니다.
  • 각 태스크를 별도의 스레드에서 실행하여 병렬 처리를 가능하게 합니다.

5. Cursor vs. Paging for Thread Safety:

• Cursor 기법:
  • JpaCursorItemReader와 같이 Cursor 기반의 ItemReader는 멀티스레드에서 안전하지 않습니다.
  • Paging 기법이나 synchronized 선언을 통해 순차적으로 수행되도록 보장되어야 합니다.
• Paging 기법:
  • JpaPagingItemReader와 같이 Paging 기반의 ItemReader는 여러 스레드 간에 안전하게 동작합니다.
  • 페이지 단위로 데이터를 읽어오기 때문에 각 페이지가 독립적인 트랜잭션으로 처리되어 스레드 간 충돌이 발생하지 않습니다.

요약:

• Chunk-oriented Processing은 대용량 데이터를 효율적으로 처리하는 데에 사용되며, 병렬 처리가 가능합니다.
• Tasklet-oriented Processing은 각 Tasklet이 독립적인 트랜잭션을 가지므로 주로 단일 작업이 필요한 경우에 사용됩니다.
• SynchronizedItemStreamReader는 멀티스레드에서 안전한 ItemStreamReader를 제공합니다.
• SimpleAsyncTaskExecutor는 간단한 비동기 태스크 실행자로, 멀티스레드 환경에서 병렬 처리를 가능하게 합니다.
• Cursor 기반 ItemReader는 멀티스레드에서 안전하지 않으며, Paging 기법이나 synchronized를 통해 스레드 간 충돌을 방지할 수 있습니다.

알림 보내기 Writer

@Slf4j
@Component
@RequiredArgsConstructor
@Transactional
public class SendNotificationItemWriter implements ItemWriter<Notification> {

    private final KafkaProducer kafkaProducer;
    private final MessageService messageService;
    private final NotificationRepository notificationRepository;

    @Override
    public void write(List<? extends Notification> notifications) throws Exception {
        int count = 0;

        for (Notification notification : notifications) {
            log.info("SendNotificationItemWriter write 실행 {}", notification);

            //문자 메시지 알림
            messageService.sendMessage(notification);

            //sse 알림
            kafkaProducer.send(new SseDto(notification.getUser().getUserId(), notification.getNotificationId()));
            count++;

            notification.updateSent();
            notificationRepository.save(notification);
        }
        log.info("SendNotificationItemWriter - write: 알람 {}/{}건 전송 성공", count, notifications.size());
    }

}

• 각 알림에 대해 문자 메시지 알림(messageService.sendMessage)과 SSE 알림(kafkaProducer.send)을 보냅니다.
• 알림을 성공적으로 보낸 경우, 해당 알림의 상태를 업데이트하고, 데이터베이스에 저장합니다.

문제해결(LazyInitializationException)

notification.getUser().getUserId()를 호출할 때 LazyInitializationException이 발생했습니다.

 
이 에러는 JPA의 지연 로딩이 활성화되어 있는 상태에서, 세션이 종료된 후에 지연로딩이 일어나려고 할 때 발생합니다.
 

fetch join을 사용하여 user를 한 번에 조회하여 해결하였습니다.

KafkaProducerConfig

@Configuration
public class KafkaProducerConfig {

    @Value("${kafka.server}")
    private String BOOTSTRAP_SERVER;

    @Bean
    public ProducerFactory<String, SseDto> newProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, SseDto> newKafkaTemplate() {
        return new KafkaTemplate<>(newProducerFactory());
    }
}

배치 서버에서 백엔드 서버로 알림 이벤트를 보내기 위해 KafkaProducerConfig를 작성합니다.

KafkaProducer

@Service
@RequiredArgsConstructor
public class KafkaProducer {

    private static final String TOPIC_NAME = "sse";

    private final KafkaTemplate<String, SseDto> kafkaTemplate;

    public void send(SseDto message) {
        kafkaTemplate.send(TOPIC_NAME, message);
    }
}

KafkaProducer가 정의한 토픽에 SseDto 메시지를 전송하는 역할을 합니다.

4. 더 알아보기

Q: 스프링 배치(Spring Batch)란 무엇인가요?
A: 스프링 배치는 대용량 데이터 처리를 위한 오픈 소스 배치 프레임워크로, 대규모 데이터 처리를 위한 일련의 작업을 지원하고 관리합니다.
 
Q: 스프링 배치의 주요 특징은 무엇인가요?
A: 주요 특징으로는 확장성, 병렬 처리, 실패 복구, 트랜잭션 관리, 재시작 기능, 간편한 설정 및 확장성이 있습니다.
 
Q: 스프링 배치와 스프링 프레임워크의 차이는 무엇인가요?
A: 스프링 배치는 대용량 데이터 처리에 특화된 프레임워크로, 스프링 프레임워크와 독립적으로 사용될 수 있습니다. 스프링 배치는 주로 배치 작업에 특화된 구조와 기능을 제공합니다.
 
Q: 스프링 배치의 핵심 컴포넌트에는 어떤 것들이 있나요?
A: 스프링 배치의 핵심 컴포넌트로는 Job, Step, ItemReader, ItemProcessor, ItemWriter 등이 있습니다.
 
Q: 스프링 배치에서 Job과 Step의 차이는 무엇인가요?
A: Job은 하나의 배치 작업 단위를 나타내며, 여러 개의 Step으로 구성됩니다. Step은 각각의 처리 단계를 나타내며, ItemReader, ItemProcessor, ItemWriter로 구성됩니다.
 
Q: 스프링 배치의 Job을 어떻게 실행하나요?
A: Job은 JobLauncher를 사용하여 실행됩니다. 스프링 배치는 커맨드 라인, 스케줄러, 웹 애플리케이션 등 다양한 방식으로 Job을 실행할 수 있습니다.
 
Q: 스프링 배치에서 ItemReader, ItemProcessor, ItemWriter의 역할은 무엇인가요?
A: ItemReader는 데이터를 읽어오는 역할, ItemProcessor는 읽어온 데이터를 가공하는 역할, ItemWriter는 가공된 데이터를 저장하는 역할을 합니다.
 
Q: 스프링 배치에서 Chunk 지향 처리는 무엇이고 어떤 장점이 있나요?
A: Chunk 지향 처리는 트랜잭션 범위 내에서 정해진 크기의 데이터를 처리하는 방식입니다. 이를 통해 대용량 데이터를 작은 덩어리로 나누어 효율적으로 처리할 수 있으며, 실패 시 해당 청크만 재시도할 수 있습니다.
 
Q: 스프링 배치에서 실패 복구는 어떻게 이루어지나요?
A: 스프링 배치는 실패한 Step 또는 Job을 기록하고, 재시작 기능을 통해 중단된 지점부터 다시 실행할 수 있습니다.
 
Q: SSE (Server-Sent Events)가 무엇인가요?
A: SSE는 서버에서 클라이언트로 실시간 업데이트를 전송하기 위한 웹 기술입니다. 이는 단방향 통신 방식으로, 서버에서 클라이언트로 푸시(Push) 메시지를 보낼 수 있습니다. 주로 실시간 업데이트가 필요한 웹 애플리케이션에서 사용됩니다.
 
Q: SSE와 웹소켓(WebSocket)의 차이는 무엇인가요?
A: SSE는 단방향 통신으로, 서버에서 클라이언트로만 데이터를 전송할 수 있습니다. 반면에 웹소켓은 양방향 통신을 지원하며, 클라이언트와 서버 간에 양방향으로 데이터를 주고받을 수 있습니다. SSE는 HTTP 기반의 프로토콜을 사용하고, 웹소켓은 별도의 프로토콜을 사용합니다.
 
Q: SSE의 작동 방식을 설명해주세요.
A: 클라이언트가 서버에 SSE 연결을 요청하면, 서버는 연결을 열고 해당 연결을 통해 이벤트 스트림을 전송합니다. 서버에서 이벤트가 발생하면, 이벤트 스트림을 통해 클라이언트로 데이터를 전송합니다. 클라이언트는 이러한 이벤트를 수신하고 처리할 수 있습니다.

참고

• 10개 프로젝트로 완성하는 백엔드 웹개발(Java/Spring) 강의를 수강하였습니다.
  • Part 5. PT 이용권 관리 서비스
  • Part 8. 대규모 트래픽을 고려한 간단한 SNS 서비스
• 이후 인프런의 스프링 배치 강의를 수강하고 있으며, 실전 카프카 개발부터 운영까지 책을 읽어보려고 합니다.