Streaming이란?

📌 1. Streaming이란?

Streaming은 데이터를 한 번에 모두 보내는 게 아니라, 조각(chunk) 단위로 나눠서 순차적으로 전송하는 방식이다.

 

📖 예를 들어:

• 대용량 파일 전송 시 전체를 다 들고 있지 않고 조금씩 읽고 보내는 방식
• 채팅, 로그, 실시간 알림 등 계속 이어지는 데이터를 실시간으로 전달하고 싶을 때 사용
• 나는 ai API 응답을 사용자에게 빨리 보여주기 위해 사용했다!!

---

📌 2. Reactive Stream 기본 개념

Reactive Stream은 비동기 + 논블로킹 방식으로 데이터 흐름(스트림)을 처리하는 기술이다.

Java에서는 이를 Flux, Mono 같은 클래스로 표현하며, Spring WebFlux는 이를 기반으로 작동하는 비동기 웹 프레임워크다.

---

📌 3. Mono vs Flux

Reactive Streams에서 사용되는 응답 형태(리턴 타입)이다.

구분	Mono	Flux
의미	0개 또는 1개의 데이터	0개 이상 (무제한 가능)
비유	단일 응답 (단건)	스트리밍 응답 (다건 / 실시간)
사용 예	로그인 결과, 사용자 정보 조회	검색 결과 리스트, 실시간 데이터 전송
응답 방식	단일 JSON	text/event-stream, chunked 등 가능
대표 메서드	bodyToMono(...)	bodyToFlux(...)

---

📌 4. 주요 연산자/함수 설명

함수	역할	주의 사항
subscribe()	실제 스트림 실행 (없으면 아무것도 안 일어남)	비동기 소비
block() / blockLast()	Mono/Flux 실행 및 결과 대기	동기 방식
doOnNext()	데이터 emit될 때마다 부가 작업	중간 로그 찍기 등
doOnSubscribe()	구독 시작 시 실행	흐름 트리거
doOnComplete()	스트림 정상 종료 시 실행	onError 시에는 호출 안 됨
doFinally()	스트림 종료 시 항상 실행 (성공/실패 모두)	리소스 정리 등에 유용
bodyToMono(...)	응답을 Mono(단일 데이터)로 변환	POST 요청 결과 등에 사용
bodyToFlux(...)	응답을 Flux(다건 스트림)으로 변환	스트리밍 응답 받을 때 사용
DataBufferUtils.read(...)	파일을 DataBuffer 스트림으로 읽음	서버 파일 응답에 유용
DataBufferUtils.write(...)	DataBuffer 스트림을 파일로 저장	클라이언트 파일 다운로드 시 활용

 

✅ bodyToMono(T), .bodyToFlux(T) 인자 타입

타입 예시	설명
DataBuffer.class	바이트 조각을 직접 받음 (로우 스트림) ->직접 CharsetDecoder로 디코딩해야 함
String.class	바이트를 문자로 디코딩한 문자열 받음
커스텀 DTO 클래스 (예: MyDto.class)	JSON, XML 등을 자동으로 역직렬화(deserialize) 하여 객체로 받음

.bodyToFlux(String.class)는 보통 작은 텍스트 조각이나 JSON 라인 단위 등에서 쓰고,
대용량 바이너리나 커스텀 인코딩 처리할 땐 .bodyToFlux(DataBuffer.class)로 받는 게 안전!

---

📌 5. 자주 사용되는 스트리밍 케이스

케이스	사용 방식	설명
실시간 로그, 알림	Flux + SSE(text/event-stream)	1초마다 데이터 전송
대용량 파일 다운로드	Flux<DataBuffer> → Mono<Void>	조각 단위 전송
파일 업로드	Flux<DataBuffer>	스트리밍으로 서버에 전달
단일 요청 결과	Mono	비동기 단건 응답 (예: 로그인)

 

✅ Flux 스트리밍 응답

@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> stream() {
    return Flux.interval(Duration.ofSeconds(1))
               .map(i -> "데이터 " + i)
               .take(5);
}

 

✅ WebClient로 Flux 받기

WebClient.create("http://localhost:8080")
    .get().uri("/stream")
    .accept(MediaType.TEXT_EVENT_STREAM)
    .retrieve()
    .bodyToFlux(String.class)
    .subscribe(System.out::println);

 

✅ 파일 다운로드 (DataBufferUtils 활용)

Flux<DataBuffer> fileStream = WebClient.create()
    .get().uri("/files/sample.pdf")
    .retrieve()
    .bodyToFlux(DataBuffer.class);

DataBufferUtils.write(fileStream, path, StandardOpenOption.CREATE)
    .block(); // 완료 대기

 

✅ 파일 응답 서버 코드

@GetMapping("/files/sample.pdf")
public Mono<Void> downloadFile(ServerHttpResponse response) {
    Flux<DataBuffer> fileStream = DataBufferUtils.read(filePath, response.bufferFactory(), 4096);
    response.getHeaders().setContentType(MediaType.APPLICATION_PDF);
    return response.writeWith(fileStream);
}

 

✅클라이언트 - WebClient로 파일 업로드하기

import org.springframework.core.io.FileSystemResource;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

public class FileUploadClient {
    public static void main(String[] args) {
        WebClient webClient = WebClient.create("http://localhost:8080");

        Mono<String> result = webClient.post()
                .uri("/upload")
                .contentType(MediaType.MULTIPART_FORM_DATA)
                .bodyValue(
                        MultipartBodyBuilderHelper.build("file", "C:/temp/sample.pdf")
                )
                .retrieve()
                .bodyToMono(String.class);

        result.subscribe(System.out::println);
    }
}

 

👉 MultipartBodyBuilderHelper는 아래처럼 따로 정의

import org.springframework.http.client.MultipartBodyBuilder;
import org.springframework.core.io.FileSystemResource;

public class MultipartBodyBuilderHelper {
    public static org.springframework.util.MultiValueMap<String, Object> build(String field, String filePath) {
        MultipartBodyBuilder builder = new MultipartBodyBuilder();
        builder.part(field, new FileSystemResource(filePath));
        return builder.build();
    }
}

 

 

✅서버 - WebFlux 컨트롤러에서 파일 수신 및 저장

import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.codec.multipart.FilePart;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;
import java.nio.file.Path;
import java.nio.file.Paths;

@RestController
public class FileUploadController {

    @PostMapping(value = "/upload", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Mono<String> handleUpload(@RequestPart("file") FilePart filePart) {
        Path path = Paths.get("uploaded_" + filePart.filename());

        // DataBuffer로 스트리밍 저장
        return DataBufferUtils.write(filePart.content(), path)
                .then(Mono.just("파일 업로드 성공: " + filePart.filename()));
    }
}

 

---

📌 6. 디코딩 처리

✅ 6-1. 디코딩 처리 흐름

• DataBuffer → byte[] → ByteBuffer 누적 → CharsetDecoder.decode() → CharBuffer → String

CharsetDecoder는 ByteBuffer만 입력으로 받으므로, DataBuffer를 직접 처리할 수 없다.

 

✅ 6-2. WebFlux DataBuffer 데이터처리(디코딩) 예제

• 서버는 UTF-8 등 문자 인코딩된 바이트를 보내고, 클라이언트/서버는 이 바이트를 문자열로 디코딩해야 최종 데이터로 활용할 수 있다.
• bodyToFlux(DataBuffer.class)로 받으면, doOnNext(buffer -> {//여기서 처리!!})

import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CoderResult;

import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;

import reactor.core.publisher.Flux;

public class DataBufferDecodingExample {

    // UTF-8 문자 디코더 생성
    private final CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();

    // 누적 바이트 버퍼 (입력 데이터 저장)
    private final ByteBuffer cumulativeBuffer = ByteBuffer.allocate(8192);

    // 문자 버퍼 (디코딩 결과 임시 저장)
    private final CharBuffer charBuffer = CharBuffer.allocate(8192);

    // 문자열 누적 저장용
    private final StringBuilder decodedStr = new StringBuilder();

    public void decodeDataBufferFlux(Flux<DataBuffer> dataBufferFlux) {
        dataBufferFlux
            .doOnNext(buffer -> {
                // 1️⃣ DataBuffer를 byte[]로 안전하게 복사
                byte[] newBytes = new byte[buffer.readableByteCount()];
                buffer.read(newBytes);
                DataBufferUtils.release(buffer); // 사용 후 반드시 해제!

                // 2️⃣ 누적 바이트 버퍼에 새 바이트 추가
                cumulativeBuffer.put(newBytes);

                // 3️⃣ 읽기 모드로 전환하여 디코딩 준비
                cumulativeBuffer.flip();

                // 4️⃣ 여러 상태에 대응해 디코딩 반복 수행
                while (true) {
                    CoderResult result = decoder.decode(cumulativeBuffer, charBuffer, false);
                    charBuffer.flip(); // 쓰기모드 → 읽기모드 전환

                    // 5️⃣ 디코딩된 문자들 누적 저장
                    decodedStr.append(charBuffer);

                    charBuffer.clear(); // 문자 버퍼 초기화

                    if (result.isUnderflow()) {
                        // 입력 바이트 부족 → 다음 조각 기다림
                        break;
                    } else if (result.isOverflow()) {
                        // 문자 버퍼 꽉 참 → 계속 디코딩
                        continue;
                    } else if (result.isError()) {
                        System.err.println("디코딩 에러 발생");
                        break;
                    }
                }

                // 6️⃣ 읽지 않은 바이트를 앞으로 이동 (다음 조각과 합치기 위해)
                cumulativeBuffer.compact();

                // 7️⃣(필요 시 여기서 decodedStr.toString()으로 문자열 사용 가능)
                try {
                    PrintWriter writer = response.getWriter();
                    writer.write("data: " + decodedStr.toString() + "\n\n"); // SSE 데이터 포맷
                    writer.flush(); // 즉시 전송
                    logger.debug("클라이언트 전송: {} ({} bytes)", decodedStr.toString(), decodedStr.length());
                } catch (IOException e) {
                    logger.error("클라이언트 전송 오류", e);
                    asyncContext.complete(); // 에러 시 비동기 컨텍스트 종료
                }

            })
            .subscribe();
    }
}

• 위 코드는 WebFlux 서버 혹은 클라이언트가 스트림으로 조각조각 오는 바이트 데이터를 실시간으로 안전하게 UTF-8 문자열로 변환하는 패턴이다.
  
• 실시간 AI 텍스트 응답 처리, 대용량 JSON 스트림 파싱, SSE(Server-Sent Events) 처리 등에 활용된다.
  
  

1️⃣왜 DataBuffer를 바로 디코딩하지 않고 복사하는가?

 DataBuffer는 Netty 기반의 임시 메모리 버퍼이므로

• 사용 후 반드시 DataBufferUtils.release(buffer)로 메모리를 해제해야 한다.
• 버퍼를 바로 재활용하지 않고, byte[]에 복사해 안전하게 처리해야 한다.
• DataBuffer는 재사용되는 풀(pool) 메모리라, 해제(release()) 후 직접 참조하면 오류가 발생한다.

2️⃣ 네트워크 데이터는 조각으로 들어와서, 단일 DataBuffer만 보고 디코딩하면 중간 조각이 깨져서 문자열 손상(�)이 생긴다.

• "hello "   → DataBuffer 1  
• "wor"      → DataBuffer 2  (단독 디코딩하면 깨짐)
• "ld\n"     → DataBuffer 3
• 따라서 안전하게 byte[]에 복사 후, ByteBuffer 누적 버퍼에 추가하여 문자 디코딩을 해야 한다.
  

3️⃣decoder.decode()는 출력용 문자 버퍼가 꽉 차면 Overflow 상태를 반환

• 이럴 때 계속 호출해서 남은 문자를 더 디코딩해야 함
  
  

그래서 while문으로 반복하며 출력 버퍼가 비워질 때까지 디코딩해야
현재 버퍼 내 모든 바이트를 처리 가능

---

📌 7. Streaming 요점 요약 🎯

• Flux는 스트림, Mono는 단건 응답
• subscribe()를 호출해야 데이터가 흐르기 시작함
• DataBuffer / DataBufferUtils는 바이너리 데이터 다룰 때 핵심
• WebFlux는 비동기 스트리밍 응답을 위한 프레임워크
• 스트리밍 응답은 text/event-stream 또는 chunked로 사용됨