73250


rustで下記コードのようにthread:spawnの内部でreqwestを利用するとthreadのJoinHandlerの所がコンパイルが通りません。  
Rustで上手くthread内でasync/awaitをどのように利用するのが良いでしょうか？

[code]

    use std::thread;
    use std::thread::JoinHandle;
    
    fn main() {
        let mut xxx = Xxxxx::new().run();
        xxx.stop();
    }
    
    async fn fetch_something() -> Result<(), reqwest::Error> {
        let body = reqwest::get("https://www.rust-lang.org")
            .await?
            .text()
            .await?;
        println!("{}", body);
        Ok(())
    }
    #[derive(Debug)]
    pub struct Xxxxx {
        worker: Option<JoinHandle<()>>,
    }
    
    impl Xxxxx {
        pub fn new() -> Self {
            Self {
                worker: None,
            }
        }
        pub fn run(mut self) -> Self {
            let child = thread::spawn(move || async {
                let _ = fetch_something().await;
            });
            self.worker = Some(child);
            self
        }
        pub fn stop(&mut self) {
            &self.worker.take().expect("").join().expect("");
        }
    }
    
    
[/code]


CC BY-SA 4.0
2021-01-11T15:19:22.920
73222
2021-01-12T16:20:33.687
43496
question
rust
rustのthread:spawn内でのasync/awaitの使い方


エラーになる直接の原因ですが、`spawn()`の返す`JointHandle`の型が、`Xxxxx`が期待する`JoinHandle`の型と合わないからです。前者は`JoinHandle<impl
Future>`で後者は`JoinHandle<()>`になります。

では、なぜ`spawn()`が`JoinHandle<impl
Future>`を返すのか（つまり根本原因は何か）というと、Rustのasyncブロック（`async { }`）やasync関数（`async fn()
{}`）は糖衣構文になっており、実際には戻り値型の値を直接返すのではなくて、戻り値型をfutureで包んだものを返すからです。asyncブロックなどから返されたfutureを実行して値を取り出すには、非同期ランタイムを使ってfutureを駆動する必要があります。

非同期ランタイムにはTokioやasync-stdなどいくつかありますが、今回はReqwestが直接依存しているTokioを使うのが楽でしょう。

まず`Cargo.toml`を以下のように修正します。

[code]

    [dependencies]
    reqwest = "0.11"
    tokio = { version = "1", features = ["rt"] }
    
[/code]

次に`Xxxxx`の`run()`を以下のように修正します。

[code]

    pub fn run(mut self) -> Self {
        let child = thread::spawn(move || {
            // 非同期ランタイムを作成する。current threadランタイムは
            // 現スレッド上で非同期タスクを実行するためのもの
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();
            // ランタイムで非同期タスク（asyncブロック）を実行する
            rt.block_on(async {
                let _ = fetch_something().await;
            });
        });
        self.worker = Some(child);
        self
    }
    
[/code]

これで`spawn()`が返す値が`JoinHandle<()>`になります。エラーが解消し、期待どおりに動作するようになります。

ところで、非同期ランタイムは複数の非同期タスクを並行して実行するためのものです。OSのスレッドに似たような機能を提供しますが、タスクの切り替えがスレッドよりもずっと軽量なことが特徴です。Reqwestを使うには非同期ランタイムが必要となり、それがOSスレッドよりも軽量なマルチタスクを提供するわけですから、`std::thread::spawn()`は使わずに非同期ランタイムだけを使うのがおすすめです。

そのように書き換えましょう。まず`Cargo.toml`を修正します。

[code]

    [dependencies]
    reqwest = "0.11"
    tokio = { version = "1", features = ["rt", "rt-multi-thread"] }
    
[/code]

次に`main.rs`を以下のように修正します。

[code]

    fn main() {
        // 非同期ランタイム（multi threadsランタイム）を作成する
        // このランタイムは複数のワーカースレッド（OSスレッド）を持ち、それらの上で
        // 非同期タスクを実行する
        let rt = tokio::runtime::Runtime::new().unwrap();
    
        // 非同期ランタイムで実行する
        rt.block_on(async {
            let mut xxx = Xxxxx::new().run();
            xxx.stop().await;
        });
    }
    
    async fn fetch_something() -> Result<(), reqwest::Error> { /* 変更なし */ }
    
    #[derive(Debug)]
    pub struct Xxxxx {
        // tokioのJoinHandleを使う
        worker: Option<tokio::task::JoinHandle<()>>,
    }
    
    impl Xxxxx {
        pub fn new() -> Self { /* 変更なし */ }
    
        pub fn run(mut self) -> Self {
            // std::thread::spawn()ではなく、tokio::spawnを使う
            let child = tokio::spawn(async move {
                let _ = fetch_something().await;
            });
            self.worker = Some(child);
            self
        }
    
        // async fnにする
        pub async fn stop(&mut self) {
            // join()をawaitに変更する
            &self.worker.take().expect("").await.expect("");
        }
    }
    
[/code]

これでOKです。

最後に、`main()`関数のところですが、Tokioが提供する`#[tokio::main]`マクロを使うと、非同期ランタイムを作るところのコードを自分で書かなくてよくなります（コンパイル時にコードが自動生成されます）

`Cargo.toml`を修正します。

[code]

    [dependencies]
    reqwest = "0.11"
    tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros"] }
    
[/code]

`main()`を以下のように修正します。

[code]

    #[tokio::main]
    async fn main() {
        let mut xxx = Xxxxx::new().run();
        xxx.stop().await;
    }
    
[/code]

これで完成です。

なお、Reqwestのblocking
featureをオンにすると、.async/awaitを全く使わないブロッキングなクライアントが使えるようになります。何かの理由で`std::thread::spawn()`を使いたいときは、ブロッキングクライアントを使う方がコードが短くてすむかもしれません。

  * <https://docs.rs/reqwest/0.11.0/reqwest/blocking/index.html>

なお、ブロッキングクライアントは内部的には最初のコードと同じようにTokioの非同期ランタイムを使っています。


CC BY-SA 4.0
2021-01-12T12:50:08.393
73250
2021-01-12T16:20:33.687
2021-01-12T16:20:33.687
14101
14101
73222
answer