[Spring Batch 마스터 클래스] Chapter 4: 병렬 처리와 파티셔닝으로 성능 극대화하기

// 기존 순차 처리 방식
@Bean
fun orderProcessingStep(): Step {
    return StepBuilder("orderProcessingStep", jobRepository)
        .chunk<Order, ProcessedOrder>(1000, transactionManager)
        .reader(orderReader())      // 1000만 건 순차 읽기
        .processor(orderProcessor()) // 하나씩 처리
        .writer(orderWriter())       // 1000건씩 저장
        .build()
}

@Configuration
class MultiThreadedOrderProcessingConfig(
    private val jobRepository: JobRepository,
    private val transactionManager: PlatformTransactionManager
) {
    
    companion object {
        private val log = LoggerFactory.getLogger(MultiThreadedOrderProcessingConfig::class.java)
    }
    
    @Bean
    fun taskExecutor(): TaskExecutor {
        val executor = ThreadPoolTaskExecutor()
        
        // 🎯 핵심 설정: 스레드 풀 크기
        executor.corePoolSize = 4        // 기본 스레드 4개
        executor.maxPoolSize = 8         // 최대 스레드 8개
        executor.queueCapacity = 200     // 대기 큐 크기
        executor.setThreadNamePrefix("batch-worker-")
        
        // 🛡️ 안전장치: 거부 정책
        executor.setRejectedExecutionHandler(ThreadPoolExecutor.CallerRunsPolicy())
        
        executor.initialize()
        return executor
    }
    
    @Bean
    fun multiThreadedOrderProcessingStep(): Step {
        return StepBuilder("multiThreadedOrderProcessingStep", jobRepository)
            .chunk<Order, ProcessedOrder>(1000, transactionManager)
            .reader(threadSafeOrderReader())      // ⚠️ Thread-safe Reader 필수!
            .processor(orderProcessor())
            .writer(threadSafeOrderWriter())      // ⚠️ Thread-safe Writer 필수!
            .taskExecutor(taskExecutor())         // 🚀 멀티스레드 활성화!
            .build()
    }
}

@Bean
@StepScope
fun threadSafeOrderReader(): JdbcPagingItemReader<Order> {
    return JdbcPagingItemReaderBuilder<Order>()
        .name("threadSafeOrderReader")
        .dataSource(dataSource)
        .selectClause("SELECT order_id, customer_id, order_date, amount")
        .fromClause("FROM orders")
        .whereClause("WHERE status = 'PENDING'")
        .sortKeys(mapOf("order_id" to Order.ASCENDING))  // 🎯 정렬 필수!
        .pageSize(1000)
        .saveState(false)  // 🚨 멀티스레드에서는 false로 설정!
        .rowMapper { rs, _ ->
            Order(
                id = rs.getLong("order_id"),
                customerId = rs.getLong("customer_id"),
                orderDate = rs.getDate("order_date").toLocalDate(),
                amount = rs.getBigDecimal("amount")
            )
        }
        .build()
}

@Component
class ThreadSafeOrderWriter : ItemWriter<ProcessedOrder> {
    
    private val dataSource: DataSource
    private val jdbcTemplate: JdbcTemplate
    
    // 🔒 동시성 제어를 위한 동기화
    private val writeLock = ReentrantLock()
    
    init {
        jdbcTemplate = JdbcTemplate(dataSource)
    }
    
    override fun write(items: List<ProcessedOrder>) {
        // 🛡️ 안전한 배치 처리
        writeLock.lock()
        try {
            val sql = """
                INSERT INTO processed_orders 
                (order_id, customer_id, processed_amount, status, processed_at)
                VALUES (?, ?, ?, ?, ?)
            """
            
            val batchArgs = items.map { order ->
                arrayOf(
                    order.id,
                    order.customerId, 
                    order.processedAmount,
                    order.status.name,
                    order.processedAt
                )
            }
            
            jdbcTemplate.batchUpdate(sql, batchArgs)
            log.info("🎯 Thread ${Thread.currentThread().name}: ${items.size}건 처리 완료")
            
        } finally {
            writeLock.unlock()
        }
    }
}

@Configuration
class ParallelStepsConfig(
    private val jobRepository: JobRepository
) {
    
    @Bean
    fun parallelProcessingJob(): Job {
        // 🎭 병렬 Flow 구성
        val parallelFlow = FlowBuilder<Flow>("parallelFlow")
            .split(taskExecutor())
            .add(
                orderProcessingFlow(),     // 주문 처리
                inventoryManagementFlow(), // 재고 관리  
                customerNotificationFlow() // 고객 알림
            )
            .build()
        
        return JobBuilder("parallelProcessingJob", jobRepository)
            .start(parallelFlow)
            .next(finalReportStep())  // 🎯 모든 병렬 작업 완료 후 실행
            .build()
    }
    
    @Bean
    fun orderProcessingFlow(): Flow {
        return FlowBuilder<Flow>("orderProcessingFlow")
            .start(orderDataProcessingStep())
            .next(paymentStatusUpdateStep())
            .build()
    }
    
    @Bean 
    fun inventoryManagementFlow(): Flow {
        return FlowBuilder<Flow>("inventoryManagementFlow")
            .start(inventoryUpdateStep())
            .next(lowStockAlertStep())
            .build()
    }
    
    @Bean
    fun customerNotificationFlow(): Flow {
        return FlowBuilder<Flow>("customerNotificationFlow")
            .start(emailNotificationStep())
            .next(smsNotificationStep())
            .build()
    }
}

@Configuration
class AsyncProcessingConfig {
    
    @Bean
    fun asyncTaskExecutor(): TaskExecutor {
        val executor = ThreadPoolTaskExecutor()
        executor.corePoolSize = 10
        executor.maxPoolSize = 20
        executor.queueCapacity = 500
        executor.setThreadNamePrefix("async-processor-")
        executor.initialize()
        return executor
    }
    
    @Bean
    fun asyncOrderProcessor(): AsyncItemProcessor<Order, ProcessedOrder> {
        val asyncProcessor = AsyncItemProcessor<Order, ProcessedOrder>()
        asyncProcessor.setDelegate(complexOrderProcessor()) // 실제 처리 로직
        asyncProcessor.setTaskExecutor(asyncTaskExecutor())
        return asyncProcessor
    }
    
    @Bean
    fun asyncOrderWriter(): AsyncItemWriter<ProcessedOrder> {
        val asyncWriter = AsyncItemWriter<ProcessedOrder>()
        asyncWriter.setDelegate(orderDatabaseWriter()) // 실제 저장 로직
        return asyncWriter
    }
    
    @Bean
    fun complexOrderProcessor(): ItemProcessor<Order, ProcessedOrder> {
        return ItemProcessor { order ->
            // 🐌 시간이 오래 걸리는 작업들 (외부 API 호출 등)
            val customerInfo = fetchCustomerInfo(order.customerId)      // 500ms
            val paymentInfo = validatePayment(order.paymentId)          // 300ms  
            val shippingCost = calculateShipping(order.address)         // 200ms
            
            ProcessedOrder(
                id = order.id,
                customerId = order.customerId,
                customerInfo = customerInfo,
                paymentInfo = paymentInfo,
                shippingCost = shippingCost,
                finalAmount = order.amount + shippingCost,
                processedAt = LocalDateTime.now()
            )
        }
    }
    
    @Bean
    fun asyncProcessingStep(): Step {
        return StepBuilder("asyncProcessingStep", jobRepository)
            .chunk<Order, Future<ProcessedOrder>>(100, transactionManager)
            .reader(orderReader())
            .processor(asyncOrderProcessor())      // 🚀 비동기 처리!
            .writer(asyncOrderWriter())            // 🚀 비동기 저장!
            .build()
    }
}

@Component
class DateRangePartitioner : Partitioner {
    
    companion object {
        private val log = LoggerFactory.getLogger(DateRangePartitioner::class.java)
    }
    
    override fun partition(gridSize: Int): Map<String, ExecutionContext> {
        val partitions = mutableMapOf<String, ExecutionContext>()
        
        // 🗓️ 2024년 1년 데이터를 gridSize만큼 분할
        val startDate = LocalDate.of(2024, 1, 1)
        val endDate = LocalDate.of(2024, 12, 31)
        val totalDays = ChronoUnit.DAYS.between(startDate, endDate)
        val daysPerPartition = totalDays / gridSize
        
        for (i in 0 until gridSize) {
            val partitionStart = startDate.plusDays(i * daysPerPartition)
            val partitionEnd = if (i == gridSize - 1) {
                endDate  // 마지막 파티션은 남은 모든 날짜 포함
            } else {
                startDate.plusDays((i + 1) * daysPerPartition - 1)
            }
            
            val context = ExecutionContext()
            context.put("startDate", partitionStart.toString())
            context.put("endDate", partitionEnd.toString())
            context.put("partitionNumber", i + 1)
            
            partitions["partition$i"] = context
            
            log.info("🎯 Partition $i 생성: $partitionStart ~ $partitionEnd")
        }
        
        log.info("🚀 총 ${partitions.size}개 파티션 생성 완료!")
        return partitions
    }
}

@Component
@StepScope
class PartitionedOrderItemReader(
    private val dataSource: DataSource,
    @Value("#{stepExecutionContext['startDate']}") private val startDate: String,
    @Value("#{stepExecutionContext['endDate']}") private val endDate: String,
    @Value("#{stepExecutionContext['partitionNumber']}") private val partitionNumber: Int
) : ItemReader<Order> {
    
    private lateinit var delegate: JdbcCursorItemReader<Order>
    
    companion object {
        private val log = LoggerFactory.getLogger(PartitionedOrderItemReader::class.java)
    }
    
    @PostConstruct
    fun init() {
        log.info("🎯 Partition $partitionNumber Reader 초기화: $startDate ~ $endDate")
        
        delegate = JdbcCursorItemReaderBuilder<Order>()
            .name("partitionedOrderReader-$partitionNumber")
            .dataSource(dataSource)
            .sql("""
                SELECT order_id, customer_id, order_date, amount, status
                FROM orders 
                WHERE order_date >= ? AND order_date <= ?
                  AND status = 'PENDING'
                ORDER BY order_id
            """)
            .preparedStatementSetter { ps ->
                ps.setDate(1, Date.valueOf(startDate))
                ps.setDate(2, Date.valueOf(endDate))
            }
            .rowMapper { rs, _ ->
                Order(
                    id = rs.getLong("order_id"),
                    customerId = rs.getLong("customer_id"),
                    orderDate = rs.getDate("order_date").toLocalDate(),
                    amount = rs.getBigDecimal("amount"),
                    status = OrderStatus.valueOf(rs.getString("status"))
                )
            }
            .build()
            
        delegate.afterPropertiesSet()
    }
    
    override fun read(): Order? {
        return delegate.read()
    }
}

@Configuration  
class PartitionedJobConfig(
    private val jobRepository: JobRepository,
    private val transactionManager: PlatformTransactionManager,
    private val dateRangePartitioner: DateRangePartitioner
) {
    
    @Bean
    fun partitionedOrderProcessingJob(): Job {
        return JobBuilder("partitionedOrderProcessingJob", jobRepository)
            .incrementer(RunIdIncrementer())
            .start(masterStep())
            .listener(partitionedJobListener())
            .build()
    }
    
    @Bean
    fun masterStep(): Step {
        return StepBuilder("masterStep", jobRepository)
            .partitioner("slaveStep", dateRangePartitioner)  // 🎯 파티셔너 설정
            .step(slaveStep())                                // 🎯 슬레이브 스텝 정의
            .gridSize(8)                                      // 🎯 파티션 개수
            .taskExecutor(partitionTaskExecutor())            // 🎯 병렬 실행용 Executor
            .build()
    }
    
    @Bean
    fun slaveStep(): Step {
        return StepBuilder("slaveStep", jobRepository)
            .chunk<Order, ProcessedOrder>(1000, transactionManager)
            .reader(partitionedOrderItemReader())
            .processor(orderProcessor())
            .writer(partitionedOrderWriter())
            .listener(slaveStepListener())
            .build()
    }
    
    @Bean
    fun partitionTaskExecutor(): TaskExecutor {
        val executor = ThreadPoolTaskExecutor()
        executor.corePoolSize = 8     // 파티션 개수와 동일하게 설정
        executor.maxPoolSize = 12     // 여유분 확보
        executor.queueCapacity = 100
        executor.setThreadNamePrefix("partition-")
        executor.initialize()
        return executor
    }
    
    @Bean
    fun partitionedJobListener(): JobExecutionListener {
        return object : JobExecutionListener {
            override fun beforeJob(jobExecution: JobExecution) {
                log.info("🚀 파티셔닝 Job 시작! GridSize: 8")
            }
            
            override fun afterJob(jobExecution: JobExecution) {
                val duration = Duration.between(
                    jobExecution.startTime,
                    jobExecution.endTime
                )
                
                if (jobExecution.status == BatchStatus.COMPLETED) {
                    log.info("✅ 파티셔닝 Job 완료! 총 소요시간: ${duration.toMinutes()}분")
                    
                    // 각 파티션별 처리 현황 출력
                    jobExecution.stepExecutions.forEach { stepExecution ->
                        if (stepExecution.stepName.startsWith("slaveStep")) {
                            log.info("📊 ${stepExecution.stepName}: " +
                                    "읽기 ${stepExecution.readCount}건, " +
                                    "쓰기 ${stepExecution.writeCount}건")
                        }
                    }
                } else {
                    log.error("❌ 파티셔닝 Job 실패!")
                }
            }
        }
    }
}

@Component
class PerformanceTuningGuide {
    
    fun calculateOptimalSettings(
        totalDataSize: Long,
        serverSpecs: ServerSpecs
    ): OptimalSettings {
        
        val cpuCores = serverSpecs.cpuCores
        val availableMemory = serverSpecs.availableMemoryGB
        
        return when {
            // 🥉 작은 데이터 (100만 건 이하)
            totalDataSize <= 1_000_000 -> {
                OptimalSettings(
                    method = "Multi-threaded",
                    threadCount = cpuCores,
                    chunkSize = 1000,
                    expectedTime = "${totalDataSize / 1000}분"
                )
            }
            
            // 🥈 중간 데이터 (100만 ~ 500만 건)
            totalDataSize <= 5_000_000 -> {
                OptimalSettings(
                    method = "Multi-threaded",
                    threadCount = cpuCores * 2,
                    chunkSize = 2000,
                    expectedTime = "${totalDataSize / 2500}분"
                )
            }
            
            // 🥇 대용량 데이터 (500만 건 이상)
            else -> {
                val optimalPartitions = minOf(cpuCores * 2, 16) // 최대 16개 파티션
                OptimalSettings(
                    method = "Partitioning",
                    partitionCount = optimalPartitions,
                    chunkSize = 1000,
                    expectedTime = "${totalDataSize / (optimalPartitions * 3000)}분"
                )
            }
        }
    }
}

data class ServerSpecs(
    val cpuCores: Int,
    val availableMemoryGB: Int
)

data class OptimalSettings(
    val method: String,
    val threadCount: Int = 0,
    val partitionCount: Int = 0,
    val chunkSize: Int,
    val expectedTime: String
)

// ❌ 위험한 코드 - Thread-Safe하지 않음
@Component
class DangerousProcessor : ItemProcessor<Order, ProcessedOrder> {
    
    private var totalAmount = BigDecimal.ZERO  // 🚨 공유 상태!
    
    override fun process(item: Order): ProcessedOrder {
        totalAmount = totalAmount.add(item.amount)  // 🚨 Race Condition!
        return ProcessedOrder(...)
    }
}

// ✅ 안전한 코드 - Thread-Safe 보장
@Component
class SafeProcessor : ItemProcessor<Order, ProcessedOrder> {
    
    private val totalAmount = AtomicReference(BigDecimal.ZERO)  // ✅ Atomic 연산
    private val processedCount = AtomicLong(0)                  // ✅ Atomic 카운터
    
    override fun process(item: Order): ProcessedOrder {
        // ✅ 안전한 업데이트
        totalAmount.updateAndGet { current -> current.add(item.amount) }
        processedCount.incrementAndGet()
        
        return ProcessedOrder(
            id = item.id,
            amount = item.amount,
            processedAt = LocalDateTime.now(),
            sequenceNumber = processedCount.get()
        )
    }
}

@Configuration
class OptimizedDataSourceConfig {
    
    @Bean
    @Primary
    fun hikariDataSource(): DataSource {
        val config = HikariConfig()
        config.jdbcUrl = "jdbc:postgresql://localhost:5432/batch_db"
        config.username = "batch_user"
        config.password = "batch_password"
        
        // 🎯 병렬 처리를 위한 최적화 설정
        val maxThreads = 16  // 최대 스레드/파티션 수
        config.maximumPoolSize = maxThreads + 5    // 여유분 확보
        config.minimumIdle = maxThreads / 2        // 최소 유지 커넥션
        
        // 🚀 성능 최적화
        config.connectionTimeout = 20000          // 20초
        config.idleTimeout = 300000               // 5분
        config.maxLifetime = 1200000              // 20분
        
        // 🔍 모니터링용 설정
        config.leakDetectionThreshold = 60000     // 1분
        
        return HikariDataSource(config)
    }
}

@Component
class MemoryOptimizedWriter : ItemWriter<ProcessedOrder> {
    
    private val batchSize = 1000
    private val buffer = mutableListOf<ProcessedOrder>()
    
    @Synchronized  // 🔒 동기화 보장
    override fun write(items: List<ProcessedOrder>) {
        buffer.addAll(items)
        
        if (buffer.size >= batchSize) {
            flushToDatabase()
        }
    }
    
    private fun flushToDatabase() {
        try {
            // 실제 DB 쓰기 작업
            jdbcTemplate.batchUpdate(sql, buffer.map { ... })
            log.info("💾 ${buffer.size}건 저장 완료")
            
        } finally {
            // 🧹 메모리 정리
            buffer.clear()
            
            // 🗑️ 메모리 사용량이 높을 때 GC 힌트
            val runtime = Runtime.getRuntime()
            val usedMemory = runtime.totalMemory() - runtime.freeMemory()
            val maxMemory = runtime.maxMemory()
            
            if (usedMemory > maxMemory * 0.8) {  // 80% 이상 사용 시
                System.gc()
                log.warn("🗑️ 메모리 사용률 높음 - GC 실행")
            }
        }
    }
}