비동기 처리 — 요청을 받고 바로 응답하지 않아도 될까

[동기 방식]
스레드 1: 요청 → DB 조회(2초 대기) → 응답  → 해제
스레드 2: 요청 → 외부 API(3초 대기) → 응답 → 해제
...
스레드 200: 요청 → 대기...
스레드 201: 거부! (스레드 풀 고갈)

[비동기 방식]
스레드 1: 요청 → 비동기 작업 시작 → 스레드 반환 (즉시 다른 요청 처리 가능)
          ... 작업 완료 ...
스레드 N: 결과 받아서 응답 전송

1. 클라이언트 요청 도착
2. 서블릿 스레드가 컨트롤러 메서드 실행
3. Callable/DeferredResult 반환
4. 서블릿 스레드 반환 (request.startAsync())
5. 별도 스레드에서 실제 작업 수행
6. 작업 완료 → DispatcherServlet에 재전달
7. 서블릿 스레드가 응답 전송

@GetMapping("/slow")
public Callable<ResponseEntity<String>> slowEndpoint() {
    return () -> {
        // 이 코드는 별도 TaskExecutor 스레드에서 실행
        Thread.sleep(3000); // 오래 걸리는 작업
        return ResponseEntity.ok("완료");
    };
}

@GetMapping("/events/{id}")
public DeferredResult<EventResponse> waitForEvent(@PathVariable Long id) {
    DeferredResult<EventResponse> deferredResult = new DeferredResult<>(30_000L); // 30초 타임아웃

    // 다른 곳(이벤트 리스너, 메시지 컨슈머 등)에서 결과를 설정할 수 있음
    eventService.registerCallback(id, event -> {
        deferredResult.setResult(new EventResponse(event));
    });

    deferredResult.onTimeout(() -> {
        deferredResult.setErrorResult(
            new ResponseStatusException(HttpStatus.REQUEST_TIMEOUT, "이벤트 대기 시간 초과")
        );
    });

    return deferredResult;
}

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean("asyncExecutor")
    public TaskExecutor asyncTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("async-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
}

@Service
public class NotificationService {

    @Async("asyncExecutor")
    public CompletableFuture<Boolean> sendEmail(String to, String subject, String body) {
        // 별도 스레드에서 실행
        boolean result = emailClient.send(to, subject, body);
        return CompletableFuture.completedFuture(result);
    }

    @Async("asyncExecutor")
    public void sendPushNotification(Long userId, String message) {
        // void 반환 — fire and forget
        pushClient.send(userId, message);
    }
}

@RestController
@RequiredArgsConstructor
public class OrderController {
    private final OrderService orderService;
    private final NotificationService notificationService;

    @PostMapping("/orders")
    public ResponseEntity<Order> createOrder(@RequestBody OrderRequest request) {
        Order order = orderService.create(request);

        // 비동기로 알림 발송 (응답에 영향 없음)
        notificationService.sendEmail(order.getUserEmail(), "주문 확인", "...");
        notificationService.sendPushNotification(order.getUserId(), "주문 완료");

        return ResponseEntity.ok(order); // 알림 완료를 기다리지 않음
    }
}

@RestController
public class NotificationController {

    private final CopyOnWriteArrayList<SseEmitter> emitters = new CopyOnWriteArrayList<>();

    // 클라이언트가 SSE 연결
    @GetMapping(value = "/notifications/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribe() {
        SseEmitter emitter = new SseEmitter(60_000L); // 60초 타임아웃

        emitters.add(emitter);

        emitter.onCompletion(() -> emitters.remove(emitter));
        emitter.onTimeout(() -> emitters.remove(emitter));
        emitter.onError(e -> emitters.remove(emitter));

        return emitter;
    }

    // 이벤트 발생 시 모든 구독자에게 전송
    public void broadcast(NotificationEvent event) {
        List<SseEmitter> deadEmitters = new ArrayList<>();

        emitters.forEach(emitter -> {
            try {
                emitter.send(SseEmitter.event()
                    .name("notification")
                    .data(event, MediaType.APPLICATION_JSON));
            } catch (IOException e) {
                deadEmitters.add(emitter);
            }
        });

        emitters.removeAll(deadEmitters);
    }
}

@GetMapping("/export/csv")
public ResponseEntity<StreamingResponseBody> exportCsv() {
    StreamingResponseBody stream = outputStream -> {
        // 서블릿 스레드를 점유하지 않고 별도 스레드에서 데이터를 쓴다
        Writer writer = new OutputStreamWriter(outputStream);
        writer.write("id,name,email\n");

        // 대용량 데이터를 청크 단위로 스트리밍
        try (Stream<User> users = userRepository.streamAll()) {
            users.forEach(user -> {
                try {
                    writer.write(user.getId() + "," + user.getName() + "," + user.getEmail() + "\n");
                    writer.flush();
                } catch (IOException e) {
                    throw new UncheckedIOException(e);

                }
            });
        }
    };

    return ResponseEntity.ok()
        .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=users.csv")
        .contentType(MediaType.parseMediaType("text/csv"))
        .body(stream);
}

// @Async 메서드의 예외 처리
@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (throwable, method, params) -> {
            log.error("비동기 메서드 예외 [{}]: {}",
                method.getName(), throwable.getMessage(), throwable);
        };
    }
}

// DeferredResult의 예외 처리
@GetMapping("/async-with-error")
public DeferredResult<String> asyncWithError() {
    DeferredResult<String> result = new DeferredResult<>(5000L);

    CompletableFuture.supplyAsync(() -> {
        // 비동기 작업
        if (someCondition) {
            throw new BusinessException(ErrorCode.INTERNAL_ERROR);

        }
        return "성공";
    }).whenComplete((value, ex) -> {
        if (ex != null) {
            result.setErrorResult(ex.getCause()); // @ExceptionHandler로 전달됨
        } else {
            result.setResult(value);
        }
    });

    return result;
}

spring:
  mvc:
    async:
      request-timeout: 30000  # 비동기 요청 타임아웃 (ms)