Rust 웹 프레임워크에서 스트리밍 응답을 사용하여 대용량 파일 및 장기 연결 효율적으로 처리하기

소개

오늘날 상호 연결된 세상에서 웹 애플리케이션은 상당한 양의 데이터를 자주 처리합니다. 수십억 바이트의 비디오 파일을 제공하는 것부터 실시간 통신 채널을 유지하는 것까지, 데이터를 효율적으로 전송하고 관리하는 능력은 무엇보다 중요합니다. 전체 응답이 전송되기 전에 버퍼링되는 기존 HTTP 요청-응답 주기는 매우 큰 파일이나 지속적인 데이터 흐름이 필요한 시나리오를 처리할 때 병목 현상이 됩니다. 이 접근 방식은 상당한 메모리를 소비하고, 지연 시간을 발생시키며, 장기 실행 작업에 대한 시간 초과를 유발할 수도 있습니다.

이것이 바로 스트리밍 응답이 강력한 솔루션으로 등장하는 곳입니다. 전체 응답이 조립될 때까지 기다리는 대신, 데이터를 사용할 수 있게 되면 청크 단위로 클라이언트에게 직접 전송합니다. 이를 통해 서버의 메모리 사용량을 줄일 뿐만 아니라 클라이언트가 더 일찍 데이터를 처리할 수 있게 되어 응답성과 사용자 경험이 크게 향상됩니다. Rust 생태계에서 Axum과 Actix Web이라는 두 가지 인기 있는 비동기 웹 프레임워크는 이러한 스트리밍 기능을 구현하기 위한 훌륭한 메커니즘을 제공합니다. 이 글에서는 이러한 프레임워크에서 스트리밍 응답을 구현하는 기술적인 측면을 깊이 파고들어 대용량 파일 제공 및 장기 연결 애플리케이션의 과제를 해결하는 방법을 보여줄 것입니다.

스트리밍 응답 및 비동기 I/O 이해

구현 세부 사항을 살펴보기 전에 관련 핵심 개념을 명확하게 이해해 봅시다.

• 스트리밍 응답: 서버가 전체 본문을 전송하기 전에 버퍼링하는 기존 HTTP 응답과 달리, 스트리밍 응답은 본문을 청크 단위로 점진적으로 보냅니다. 이를 통해 클라이언트는 전체 응답을 기다리지 않고 데이터가 도착하는 대로 데이터를 수신하고 처리할 수 있습니다. 특히 대용량 파일, 실시간 데이터 피드 또는 장기 실행 계산에 유용합니다.
• 비동기 I/O: Axum 및 Actix Web과 같은 Rust 웹 프레임워크의 핵심은 비동기 I/O입니다. 이 패러다임은 단일 스레드가 차단 없이 여러 I/O 작업(디스크에서 읽기 또는 네트워크로 데이터 전송 등)을 동시에 관리할 수 있도록 합니다. 작업이 완료될 때까지 기다리는 대신, 스레드는 다른 작업으로 전환하고 I/O가 준비되면 원래 작업을 재개할 수 있습니다. 이 논블로킹 특성은 서버가 단일 클라이언트나 느린 I/O 작업에 의해 중단되지 않고 지속적으로 데이터를 보낼 수 있으므로 효율적인 스트리밍에 중요합니다.
• tokio::fs::File 및 tokio::io::AsyncReadExt / AsyncWriteExt: 비동기 Rust 애플리케이션에서 파일 작업을 할 때 tokio::fs::File은 std::fs::File의 논블로킹 등가물입니다. 관련 트레이트인 AsyncReadExt 및 AsyncWriteExt는 Rust의 async/await 구문 및 Tokio 런타임과 완벽하게 통합되는 read 및 write와 같은 비동기 메서드를 제공합니다.
• futures::Stream: futures 크레이트의 이 트레이트는 시간이 지남에 따라 비동기적으로 생성되는 값의 시퀀스를 나타냅니다. Iterator의 비동기 버전이며 사용자 지정 스트리밍 응답을 구축하는 데 기본이 되어 데이터 청크가 생성되고 전송되는 방식을 정의할 수 있습니다.

스트리밍 응답의 원리는 간단합니다. 서버는 클라이언트와 HTTP 연결을 설정하고, 전체 응답을 한 번에 보내는 대신, 작은 데이터 청크를 지속적으로 보냅니다. 이는 종종 HTTP/1.1의 Transfer-Encoding: chunked 메커니즘을 사용하여 달성되며, 각 데이터 청크 앞에 크기가 붙습니다. Rust의 비동기 웹 프레임워크의 비동기 특성은 이를 완벽하게 보완하여 서버가 스레드를 차단하지 않고 여러 동시 스트리밍 연결을 효율적으로 관리할 수 있도록 합니다.

Axum에서 스트리밍 응답 구현

Tokio 및 Hyper를 기반으로 하는 Axum은 스트리밍을 유연하고 구성 가능하게 처리할 수 있는 방법을 제공합니다. 핵심은 http_body::Body를 구현하는 응답 본문을 반환하거나 Axum의 내장 StreamBody를 활용하는 것입니다.

예제 1: 디스크에서 대용량 파일 스트리밍

서버에 있는 대용량 비디오 파일을 제공한다고 가정해 봅시다.

use axum::{
    body::{Body, Bytes},
    extract::Path,
    http::{
        header::{CONTENT_DISPOSITION, CONTENT_TYPE},
        StatusCode,
    },
    response::{IntoResponse, Response},
    routing::get,
    Router,
};
use tokio::{fs::File, io::AsyncReadExt};
use tokio_util::io::ReaderStream;
use futures::StreamExt; // Required for .map() on ReaderStream

#[tokio::main]
async fn main() {
    // 데모를 위한 더미 대용량 파일 생성
    // 실제 애플리케이션에서는 이 파일이 이미 존재할 것입니다.
    tokio::fs::write("large_file.bin", vec![0u8; 1024 * 1024 * 50]).await.unwrap(); // 50MB 파일

    let app = Router::new()
        .route("/download/file/:filename", get(stream_file));

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000")
        .await
        .unwrap();
    println!("Listening on http://127.0.0.1:3000");
    axum::serve(listener, app).await.unwrap();
}

async fn stream_file(Path(filename): Path<String>) -> Result<Response, StatusCode> {
    let path = format!("./{{}}", filename);
    let file = match File::open(&path).await {
        Ok(file) => file,
        Err(err) => {
            eprintln!("Error opening file: {}: {{}}", path, err);
            return Err(StatusCode::NOT_FOUND);
        }
    };

    // 파일 메타데이터를 가져와 콘텐츠 길이와 수정 시간을 결정
    let metadata = file.metadata().await.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
    let file_size = metadata.len();

    // 파일에서 ReaderStream 생성
    let stream = ReaderStream::new(file);

    // Bytes 스트림을 Body로 변환
    // 스트림 항목에 실패할 수 있는 경우 여기에 오류 처리를 추가할 수 있습니다.
    let body = Body::from_stream(stream);

    // 적절한 헤더로 응답 빌드
    let response = Response::builder()
        .status(StatusCode::OK)
        .header(CONTENT_TYPE, "application/octet-stream") // 또는 mime 유형 감지
        .header(
            CONTENT_DISPOSITION,
            format!("attachment; filename=\"{{}}\"", filename),
        )
        // 대용량 파일의 경우, chunked 인코딩을 사용하면 Content-Length가 종종 생략되지만,
        // 알려진 경우 클라이언트에게 유용할 수 있습니다.
        // chunked 인코딩을 사용하지 않는 경우 Content-Length가 중요합니다. Axum/Hyper
        // 일반적으로 스트림을 사용할 때 chunked 인코딩을 자동으로 처리합니다.
        .body(body)
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(response)
}

이 예제에서:

1. tokio::fs::File::open을 사용하여 large_file.bin을 비동기적으로 엽니다.
2. tokio_util::io::ReaderStream::new(file)은 비동기 파일 리더를 Bytes 청크의 Stream으로 변환합니다.
3. Body::from_stream(stream)은 이 스트림을 가져와 Axum Body로 래핑하여 데이터를 청크 단위로 보내는 방법을 알고 있습니다.
4. 클라이언트에게 다운로드를 제안하기 위해 CONTENT_TYPE 및 CONTENT_DISPOSITION 헤더를 설정합니다.

브라우저나 curl로 http://127.0.0.1:3000/download/file/large_file.bin에 액세스하면 서버가 50MB 전체를 미리 버퍼링하지 않고 파일이 즉시 청크 단위로 다운로드되는 것을 볼 수 있습니다.

예제 2: 생성된 데이터 스트리밍(장기 연결)

때로는 동적으로 생성된 데이터를 스트리밍해야 할 수도 있습니다. 예를 들어, 장기 실행 계산이나 실시간 데이터 소스에서 가져온 데이터일 수 있습니다.

use axum::{
    body::{Body, Bytes},
    response::{IntoResponse, Response},
    routing::get,
    Router,
};
use futures::Stream;
use std::{pin::Pin, task::{Context, Poll}, time::Duration};
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/live_messages", get(stream_generated_data));

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000")
        .await
        .unwrap();
    println!("Listening on http://127.0.0.1:3000");
    axum::serve(listener, app).await.unwrap();
}

// 매초 메시지를 생성하는 사용자 지정 스트림
struct MessageStream {
    counter: usize,
}

impl Stream for MessageStream {
    type Item = Result<Bytes, &'static str>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        if self.counter >= 10 { // 10개의 메시지 후에 중지
            return Poll::Ready(None);
        }

        // 비차단 지연을 위해 Tokio의 sleep 사용
        // 이는 스트림을 비동기적이고 협력적으로 만듭니다.
        let fut = Box::pin(sleep(Duration::from_secs(1)));
        tokio::pin!(fut); // 퓨처를 스택에 고정

        match fut.poll(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(_) => {
                let message = format!("Message {{}} from server\n", self.counter);
                self.counter += 1;
                Poll::Ready(Some(Ok(Bytes::from(message))))
            }
        }
    }
}

async fn stream_generated_data() -> Response {
    let stream = MessageStream { counter: 0 };

    let body = Body::from_stream(stream);

    Response::builder()
        .status(200)
        .header("Content-Type", "text/plain")
        .body(body)
        .unwrap()
}

여기서 MessageStream은 futures::Stream을 구현하여 매초 메시지를 생성합니다. 각 메시지는 Bytes로 변환되어 클라이언트로 전송됩니다. 이는 서버 전송 이벤트(SSE)와 유사한 시나리오 또는 실시간 데이터 피드를 시뮬레이션합니다. 브라우저에서 http://127.0.0.1:3000/live_messages를 열면 메시지가 점진적으로 나타나는 것을 볼 수 있습니다.

Actix Web에서 스트리밍 응답 구현

Actix Web 역시 actix_web::web::Bytes 및 actix_web::Responder 트레이트와 actix_web::body::MessageBody를 통해 스트리밍 응답을 강력하게 지원합니다. 원시 스트림 처리를 위해 futures::Stream을 사용하는 actix_web::web::Bytes가 최선의 방법입니다.

예제 1: 디스크에서 대용량 파일 스트리밍

use actix_web::{
    get, App, HttpResponse, HttpServer, Responder,
    web,
    http::header::{ContentDisposition, DispositionType},
    body::BoxedStream, // Boxed 스트림 반환용
};
use tokio::{fs::File, io::AsyncReadExt};
use tokio_util::io::ReaderStream;
use futures::StreamExt; // ReaderStream의 .map()용

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    tokio::fs::write("large_file.bin", vec![0u8; 1024 * 1024 * 50]).await.unwrap(); // 50MB 파일

    HttpServer::new(|| {
        App::new()
            .service(download_file)
    })
    .bind("127.0.0.1:8080")? 
    .run()
    .await
}

#[get("/download/file/{filename}")]
async fn download_file(web::Path(filename): web::Path<String>) -> actix_web::Result<HttpResponse> {
    let path = format!("./{{}}", filename);
    let file = File::open(&path)
        .await
        .map_err(actix_web::error::ErrorInternalServerError)?; // tokio::io::Error를 actix_web::Error로 변환

    // 파일에서 ReaderStream 생성
    let stream = ReaderStream::new(file)
        .map(|res| res.map_err(|e| actix_web::error::ErrorInternalServerError(e))); // tokio::io::Error를 actix_web::Error로 매핑

    Ok(HttpResponse::Ok()
        .content_type("application/octet-stream")
        .insert_header(ContentDisposition::attachment(&filename))
        .streaming(stream /* 유연성을 위해 필요한 경우 BoxedStream<_, _>로 */ ))
}

Axum 예제와 유사하게:

1. 파일을 비동기적으로 엽니다.
2. tokio_util::io::ReaderStream은 파일에서 Bytes 스트림을 생성합니다.
3. .map() 호출은 여기서 Result<Bytes, tokio::io::Error> 항목을 Actix Web의 오류 처리가 예상하는 Result<Bytes, actix_web::Error>로 변환하는 데 중요합니다.
4. HttpResponse::Ok().streaming(stream)은 스트리밍 응답을 구성합니다. Actix Web은 Transfer-Encoding: chunked 헤더를 자동으로 처리합니다.

예제 2: 생성된 데이터 스트리밍(장기 연결)

use actix_web::{
    get, App, HttpResponse, HttpServer, Responder,
    http::header::ContentType,
    body::BoxedStream,
};
use futures::Stream;
use std::{pin::Pin, task::{Context, Poll}, time::Duration};
use tokio::time::sleep;

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .service(live_messages)
    })
    .bind("127.0.0.1:8080")? 
    .run()
    .await
}

struct MessageStream {
    counter: usize,
}

impl Stream for MessageStream {
    type Item = Result<actix_web::web::Bytes, &'static str>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        if self.counter >= 10 {
            return Poll::Ready(None);
        }

        let fut = Box::pin(sleep(Duration::from_secs(1)));
        tokio::pin!(fut);

        match fut.poll(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(_) => {
                let message = format!("Message {{}} from server\n", self.counter);
                self.counter += 1;
                Poll::Ready(Some(Ok(actix_web::web::Bytes::from(message))))
            }
        }
    }
}

#[get("/live_messages")]
async fn live_messages() -> HttpResponse {
    let stream = MessageStream { counter: 0 };

    HttpResponse::Ok()
        .content_type(ContentType::plaintext())
        .streaming(stream)
}

Actix Web에서 생성된 데이터에 대한 접근 방식은 Axum과 매우 유사합니다. MessageStream에 대해 futures::Stream을 구현하여 항목 유형이 Result<actix_web::web::Bytes, E>이고 E가 actix_web::Error로 변환될 수 있는 오류 유형인지 확인합니다. 그런 다음 HttpResponse::streaming은 이 스트림을 가져옵니다.

애플리케이션 시나리오

스트리밍 응답은 매우 다재다능하며 다양한 시나리오에서 사용됩니다.

• 대용량 미디어 파일 제공: 비디오, 고해상도 이미지 및 대용량 아카이브를 직접 스트리밍하여 서버 메모리 사용량을 줄이고 클라이언트가 전체 파일을 다운로드하기 전에 재생하거나 처리를 시작할 수 있도록 합니다.
• 실시간 데이터 피드(서버 전송 이벤트 - SSE): 뉴스 업데이트, 주가, 채팅 메시지 또는 IoT 센서 데이터는 서버가 이벤트가 발생하는 대로 스트리밍하는 장기 실행 HTTP 연결을 통해 클라이언트로 푸시될 수 있습니다.
• 장기 실행 API 작업: API 호출이 결과를 계산하는 데 상당한 시간이 걸리는 경우, 스트리밍을 통해 서버는 완료될 때까지 연결을 유지하는 대신 부분 결과 또는 진행률 업데이트를 클라이언트로 보낼 수 있습니다.
• 백업 및 복원 서비스: 백업 솔루션을 위한 스트리밍 파일 업로드 또는 다운로드는 서버 메모리가 고갈되지 않고 임의 크기의 파일을 처리할 수 있습니다.
• 로그 테일 보기: 웹 인터페이스는 명령줄에서 tail -f가 작동하는 방식과 유사하게 서버에서 실시간 로그를 스트리밍할 수 있습니다.

결론

Axum 및 Actix Web과 같은 Rust의 비동기 웹 프레임워크에서 스트리밍 응답은 대용량 파일 처리 및 장기 연결 유지를 위한 강력하고 효율적인 메커니즘을 제공합니다. 비동기 I/O의 논블로킹 특성과 futures::Stream 트레이트를 활용함으로써 개발자는 메모리 사용량을 줄이고 지연 시간을 개선하며 전반적인 사용자 경험을 향상시키는 데이터를 점진적으로 제공하는 반응성이 뛰어나고 확장 가능한 애플리케이션을 구축할 수 있습니다. 이 접근 방식은 데이터 집약적이고 실시간 상호 작용 요구 사항을 수용할 수 있는 최신 웹 서비스를 구축하는 데 기본이 됩니다. 스트리밍 구현은 Rust에서 고성능 및 리소스 효율적인 웹 애플리케이션의 핵심입니다.