TIL

ItemWriter와 ItemReader 통합

CompositeItemReader

• 여러 ItemReader들을 순차적으로 실행
• ItemReader를 List로 전달하면 이를 순서대로 실행한다.
  • 이전 ItemReader가 더 이상 읽을 게 없다고 판단되면 null을 반환

List<ItemStreamReader<Customer>> readers = List.of(
    shard1ItemReader,  
    shard2ItemREader   
);

CompositeItemReader<Customer> compositeReader = new CompositeItemReader<>(readers);

• 데이터베이스가 여러 샤드로 분산되어 있을 때 특히 유용하다.
  • 샤드 관리 없이 각 샤드(datasource)를 바라보는 ItemReader들만 준비하면 된다.

CompositeItemWriter

• CompositeItemWriter는 여러 ItemWriter들에게 같은 데이터를 전달한다.
  • 모든 위임 대상들이 차례대로 write()를 호출

public void write(Chunk<? extends T> chunk) throws Exception {
    for (ItemWriter<? super T> writer : delegates) {
       writer.write(chunk);
    }
}

• 구성도 단순하다.

CompositeItemWriter<Hacker> writer = new CompositeItemWriter<>(
   List.of(
       firstWriter,
       secondWriter,
       thirdWriter
   )
);

• 하나의 데이터를 여러 시스템에 동시에 써야하는 경우 유용하다.
  • ex) 동일한 데이터를 RDB와 NoSQL에 각각 저장해야 하는 경우
• CompositeItemWriter 사용에선 FlatFileItemWriter와 MongoItemWriter의 지연 쓰기가 빛을 발한다.
  • FlatFileItemWriter와 MongoItemWriter는 beforeCommit() 시점까지 실제 쓰기를 미룬다.
  • ex) FlatFileItemWriter, JdbcBatchItemWriter로 CompositeItemWriter를 구성한 경우
    • JdbcBatchItemWriter에서 예외가 발생한 경우 RDB에선 자연스레 롤백이 일어난다.
    • 이 때 이미 FlatFileItemWriter.write()가 실행되었고, 즉시 데이터를 썼다면 파일의 데이터와 RDB 데이터 싱크는 어긋난다.
    • 쓰기 지연 덕분에 데이터를 쓰지 않고 넘어갈 수 있다.

ClassifierCompositeItemWriter

• CompositeItemWriter에선 모든 ItemWriter들이 같은 데이터를 처리해야 한다.
• ClassifierCompositeItemWriter는 여러 데이터들을 처리할 수 있다.
  • Spring의 Classifier를 사용해 데이터를 분기
  • write() 메서드에 청크가 전달되면 classify() 메서드가 동작
  • 각 데이터 처리 방식에 따라 위임 ItemWriter가 선택된다.

// ClassifierCompositeItemWriter 내부의 classifier 필드
private Classifier<T, ItemWriter<? super T>> classifier = new ClassifierSupport<>(null);

구성 방법

• ClassifierCompositeItemWriter 구성하는 두 가지 방법
  • ClassifierCompositeItemWriter 객체 생성 후 setClassifer()로 분류기 설정
  • ClassifierCompositeItemWriterBuilder.classifier()로 한 번의 설정

예제 코드

    class SystemLogClassifier(
        private val criticalWriter: ItemWriter<SystemLog>,
        private val normalWriter: ItemWriter<SystemLog>,
    ) : Classifier<SystemLog, ItemWriter<in SystemLog>> {
        companion object {
            const val CRITICAL_CPU_THRESHOLD = 90
            const val CRITICAL_MEMORY_THRESHOLD = 1024L * 1024 * 1024 // 1GB
        }

        override fun classify(log: SystemLog): ItemWriter<in SystemLog> =
            if (isCritical(log)) {
                criticalWriter
            } else {
                normalWriter
            }

        private fun isCritical(log: SystemLog): Boolean =
            log.type == "CRITICAL" ||
                log.cpuUsage >= CRITICAL_CPU_THRESHOLD ||
                log.memoryUsage >= CRITICAL_MEMORY_THRESHOLD
    }

    @Bean
    fun classifierWriter(): ClassifierCompositeItemWriter<SystemLog> =
        ClassifierCompositeItemWriter<SystemLog>()
            .apply {
                setClassifier(
                    SystemLogClassifier(
                        criticalLogWriter(),
                        normalLogWriter(),
                    ),
                )
            }

    @Bean
    fun normalLogWriter(): ItemWriter<SystemLog> =
        ItemWriter { items ->
            log.info { "✅NormalLogWriter: 일반 로그 처리 중... 대충 파일에 출력하거나 하자.." }
            items.forEach { item ->
                log.info { "✅일반 처리: $item" }
            }
        }

    @Bean
    fun criticalLogWriter(): ItemWriter<SystemLog> =
        ItemWriter { items ->
            log.info { "🚨CriticalLogWriter: 치명적 시스템 로그 감지! 즉시 처리 시작!" }
            items.forEach { item ->
                // 실제 운영에선 여기서 슬랙 혹은 이메일 발송
                log.info { "🚨긴급 처리: $item" }
            }
        }