문서가격블로그
문의하기
로그인무료로 가입하기
Back to Tech Articles

Rust Webフレームワークにおけるストリーミングレスポンスの効率的な大容量ファイルと長時間接続の処理

7월 22, 2025
# Misc

Grace Collins

Solutions Engineer · Leapcell

이 게시물은 English, 한국어 및 Deutsch에서도 볼 수 있습니다.
Cover of "Rust Webフレームワークにおけるストリーミングレスポンスの効率的な大容量ファイルと長時間接続の処理"

はじめに

今日の相互接続された世界では、Webアプリケーションはしばしば大量のデータを扱います。数ギガバイトのビデオファイルを配信したり、リアルタイム通信チャネルを維持したりする場合でも、効率的なデータ転送と管理能力が不可欠です。応答全体が送信される前にバッファリングされる従来のHTTPリクエスト・レスポンスサイクルは、非常に大きなファイルを扱ったり、連続したデータフローを必要としたりするシナリオではボトルネックになります。このアプローチは、大量のメモリを消費し、遅延を発生させ、長時間実行される操作のタイムアウトにつながる可能性さえあります。

ここで、ストリーミングレスポンスが強力なソリューションとして登場します。応答全体が組み立てられるのを待つ代わりに、データは利用可能になり次第、チャンクごとにクライアントに直接送信されます。これにより、サーバーのメモリフットプリントが削減されるだけでなく、クライアントはより早くデータの処理を開始できるようになり、応答性とユーザーエクスペリエンスが大幅に向上します。Rustエコシステムでは、AxumとActix Webという2つの人気のある非同期Webフレームワークが、このようなストリーミング機能を実装するための優れたメカニズムを提供しています。この記事では、これらのフレームワークでストリーミングレスポンスを実装する上での技術的な詳細を掘り下げ、大容量ファイルの配信や長時間接続アプリケーションの課題に対処する方法をデモンストレーションします。

ストリーミングレスポンスと非同期I/Oの理解

実装の詳細に入る前に、関連するコアコンセプトを明確に理解しましょう。

  • ストリーミングレスポンス: サーバーが応答全体をバッファリングしてから送信する従来のHTTPレスポンスとは異なり、ストリーミングレスポンスはチャンクで段階的に本文を送信します。これにより、クライアントは完全な応答を待たずに、データが到着次第処理することができます。これは、大きなファイル、リアルタイムデータフィード、または長時間実行される計算に特に役立ちます。
  • 非同期I/O: AxumやActix WebのようなRustのWebフレームワークの核となるのは非同期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のWebフレームワークの非同期性質は、これを完璧に補完し、サーバーがスレッドを解放することなく、複数の同時ストリーミング接続を効率的に管理できるようにします。

Axumでのストリーミングレスポンスの実装

TokioとHyper上に構築されたAxumは、ストリーミングを処理するための柔軟で composable な方法を提供します。鍵となるのは、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;
use tokio::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 file

    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),
        )
        // 大容量ファイルの場合、チャンクエンコーディングではContent-Lengthは省略されることが多いですが、
        // 知っている場合はクライアントに役立ちます。
        // チャンクエンコーディングを使用しない場合、Content-Lengthは不可欠です。Axum/Hyper は通常、ストリームを使用する際にチャンクエンコーディングを自動的に処理します。
        .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();
}

// 1秒ごとにメッセージを生成するカスタムストリーム
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); // Futureをスタックにピン留め

        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 を実装して、1秒ごとにメッセージを生成します。各メッセージは 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, // ボックス化されたストリームを返すため
];
use tokio::fs::File;
use tokio::io::AsyncReadExt;
use tokio_util::io::ReaderStream;
use futures::StreamExt; // .map() for ReaderStream

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

    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呼び出しに結果の計算にかなりの時間がかかる場合、ストリーミングにより、サーバーは完了まで接続を維持するのではなく、部分的な結果や進捗更新をクライアントに送信できます。
  • バックアップおよび復元サービス: バックアップソリューション用のストリーミングファイルアップロードまたはダウンロードは、サーバーメモリを使い果たすことなく、任意のサイズのファイルを処理できます。
  • ログテールビュー: Webインターフェースは、コマンドラインの tail -f のように、サーバーからライブログをストリーミングできます。

結論

AxumやActix WebのようなRustの非同期Webフレームワークにおけるストリーミングレスポンスは、大容量ファイルや長時間接続を処理するための強力で効率的なメカニズムを提供します。非同期I/Oのノンブロッキング性質と futures::Stream トレイトを活用することで、開発者はメモリフットプリントを削減し、遅延を改善し、全体的なユーザーエクスペリエンスを向上させる、増分的にデータを配信する応答性とスケーラブルなアプリケーションを構築できます。このアプローチは、データ集約型およびリアルタイムインタラクションの要求を円滑に処理できる最新のWebサービスを構築するための基盤となります。ストリーミングの実装は、Rustにおける高性能でリソース効率の高いWebアプリケーションの基盤です。

Share this article

HackerNews

More Posts from Leapcell

Popular Posts