Spring Batch 구현 - 4

 
♧ 전체 코드 : https://github.com/woodisco/pass-batch

GitHub - woodisco/pass-batch

 

수업 전 알람 배치

① SendNotificationBeforeClassJobConfig 작성

@Bean
public Step sendNotificationStep(JobRepository jobRepository,
                                 PlatformTransactionManager transactionManager,) {
    return new StepBuilder("sendNotificationStep", jobRepository)
            .<NotificationEntity, NotificationEntity>chunk(CHUNK_SIZE, transactionManager)
            .reader(sendNotificationItemReader())
            .writer(sendNotificationItemWriter)
            .taskExecutor(new SimpleAsyncTaskExecutor())
            .build();
}

// SynchronizedItemStreamReader: multi-thread 환경에서 reader와 writer는 thread-safe 해야합니다.
// Cursor 기법의 ItemReader는 thread-safe 하지 않아 Paging 기법을 사용하거나 synchronized 를 선언하여 순차적으로 수행해야합니다.
@Bean
public SynchronizedItemStreamReader<NotificationEntity> sendNotificationItemReader() {
    JpaCursorItemReader<NotificationEntity> itemReader = new JpaCursorItemReaderBuilder<NotificationEntity>()
            .name("sendNotificationItemReader")
            .entityManagerFactory(entityManagerFactory)
            .queryString("select n from NotificationEntity n where n.event = :event and n.sent = :sent")
            .parameterValues(Map.of("event", NotificationEvent.BEFORE_CLASS, "sent", false))
            .build();

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

▷ .taskExecutor(new SimpleAsyncTaskExecutor()) :
.taskExecutor(new SimpleAsyncTaskExecutor()) 코드는 Spring Batch에서 병렬 처리를 구현하기 위해 사용됩니다. 이 코드는 작업(Chunk)을 실행할 때 사용할 TaskExecutor를 설정합니다. 여기서 SimpleAsyncTaskExecutor는 가장 간단한 형태의 TaskExecutor 중 하나입니다.

SimpleAsyncTaskExecutor는 각각의 작업을 새로운 스레드에서 실행하고, 해당 스레드들을 병렬로 동시에 실행하여 처리합니다. 이는 작업을 여러 스레드에서 동시에 실행하여 처리 속도를 향상할 수 있도록 해줍니다.

즉, .taskExecutor(new SimpleAsyncTaskExecutor()) 코드는 Step에서 병렬 처리를 위해 간단한 TaskExecutor를 설정하는 것입니다. 이렇게 하면 작업이 동시에 여러 스레드에서 실행되어 처리 속도를 향상할 수 있습니다.

▷ SynchronizedItemStreamReader :
데이터 소스에서 아이템을 읽을 때 스레드 안전성을 보장하기 위해 사용됩니다, 특히 다중 스레드 환경에서. 사용 이유를 살펴보겠습니다:
・ 스레드 안전성: 다중 스레드 환경에서 여러 스레드가 동시에 아이템을 읽으려고 할 수 있습니다. 동기화가 없으면 이는 경쟁 상태와 데이터 손상을 유발할 수 있습니다. SynchronizedItemStreamReader를 사용하면 기본 아이템 리더(itemReader 이 경우)에 대한 접근이 동기화되어, 한 번에 하나의 스레드만 접근할 수 있도록 보장됩니다. 이는 데이터 손상을 방지하고 일관된 동작을 보장합니다.
・ 병렬 처리: 일괄 처리 시나리오에서 성능을 향상하기 위해 여러 스레드를 통해 아이템을 동시에 처리하는 것이 일반적입니다. SynchronizedItemStreamReader는 여러 스레드가 동일한 아이템 리더에 안전하게 접근할 수 있도록 하여, 데이터 무결성 문제없이 병렬 처리를 가능하게 합니다.
・ Spring Batch와 호환성: Spring Batch는 일괄 처리를 위한 여러 구성 요소를 제공합니다. 이 중 SynchronizedItemStreamReader는 Spring Batch 프레임워크의 일부이며, 다른 Spring Batch 구성 요소와 원활하게 작동하도록 설계되었습니다. Spring Batch 작업 내에서 아이템 읽기 작업에 동기화를 추가하는 간편한 방법을 제공합니다.

요약하면, SynchronizedItemStreamReader는 다중 스레드 일괄 처리 환경에서 아이템 읽기 작업에 스레드 안전성을 추가하여 데이터 무결성을 보장하고 병렬 처리를 가능하게 합니다.

▷ .delegate(itemReader) :
빌더에게 "이 itemReader를 실제 데이터를 읽을 때 사용해"라고 말하는 것과 같습니다. 이는 동기화된 아이템 스트림 리더가 데이터를 읽는 작업을 수행하기 위해 실제 작업자(여기서는 itemReader)를 설정합니다.

 
② NotificationModelMapper 작성
③ LocalDateTimeUtils 작성
④ SendNotificationItemWriter 작성
⑤ KakaoTalkMessageAdapter 작성
https://developers.kakao.com/docs/latest/en/message/rest-api

Kakao Developers

@Service
public class KakaoTalkMessageAdapter {
    private final WebClient webClient;

    public KakaoTalkMessageAdapter(KakaoTalkMessageConfig config) {
        webClient = WebClient.builder()
                .baseUrl(config.getHost())
                .defaultHeaders(h -> {
                    h.setBearerAuth(config.getToken());
                    h.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
                }).build();
    }

    public boolean sendKakaoTalkMessage(final String uuid, final String text) {
        KakaoTalkMessageResponse response = webClient.post().uri("/v1/api/talk/friends/message/default/send")
                .body(BodyInserters.fromValue(new KakaoTalkMessageRequest(uuid, text)))
                .retrieve()
                .bodyToMono(KakaoTalkMessageResponse.class)
                .block();
        if (response == null || response.getSuccessfulReceiverUuids() == null) {
            return false;

        }
        return response.getSuccessfulReceiverUuids().size() > 0;
    }
}

▷ WebClient :
Spring Framework에서 제공하는 비동기적인 HTTP 클라이언트입니다. 이것은 HTTP 리소스를 소비하는 데 사용됩니다. 기존의 RestTemplate보다 더욱 유연하고 효율적으로 작동합니다. WebClient는 다양한 HTTP 작업을 수행할 수 있습니다. GET, POST, PUT, DELETE와 같은 HTTP 메서드를 지원하며, JSON, XML, FormData와 같은 다양한 형식으로 데이터를 전송할 수 있습니다. 또한 HTTP 요청 및 응답에 대한 다양한 설정도 제공합니다. WebClient는 RESTful 웹 서비스와 상호작용하는 데 매우 유용합니다. 이를 통해 원격 API 호출, 마이크로서비스 간 통신 등 다양한 시나리오에서 효과적으로 작업할 수 있습니다.

▷ .retrieve().bodyToMono(KakaoTalkMessageResponse.class).block(); :
.retrieve()는 WebClient로부터 응답을 받기 위한 메서드입니다. 이 메서드를 호출하면 HTTP 요청을 보내고, 서버로부터 응답을 받아옵니다.
.bodyToMono(KakaoTalkMessageResponse.class)는 받아온 응답 데이터를 원하는 형식으로 변환하는 부분입니다. 여기서는 받아온 응답 데이터를 KakaoTalkMessageResponse 클래스의 객체로 변환하고 있습니다. 이 클래스는 서버에서 받은 응답 데이터를 자바 객체로 매핑하기 위해 사용됩니다.
.block()은 Mono 형식의 데이터를 실제 값으로 가져오는 역할을 합니다. Mono는 비동기적으로 값을 처리하기 위한 리액티브 타입 중 하나인데,. block() 메서드를 호출하면 해당 Mono가 가지고 있는 값을 가져와서 반환합니다. 이때 블로킹 방식으로 동작하므로 주의가 필요합니다.

쉽게 말하자면, 이 코드는 WebClient로 HTTP 요청을 보내고, 그에 대한 응답을 받아와서 KakaoTalkMessageResponse 클래스의 객체로 변환한 뒤에 그 값을 반환하는 것입니다.

 
⑥ KakaoTalkMessageRequest, KakaoTalkMessageResponse 작성
⑦ KakaoTalkMessageConfig 작성
 
출처 : 패스트캠퍼스 10개 프로젝트로 완성하는 백엔드 웹개발(Java/Spring) 초격차 패키지 Online