如何停止从 tokio::io::lines 流中读取？

Question

我想终止从tokio::io::lines 流中读取。我将它与 oneshot 未来合并并终止了它，但 tokio::run 仍在工作。

use futures::{sync::oneshot, *}; // 0.1.27
use std::{io::BufReader, time::Duration};
use tokio::prelude::*; // 0.1.21

fn main() {
    let (tx, rx) = oneshot::channel::<()>();
    let lines = tokio::io::lines(BufReader::new(tokio::io::stdin()));
    let lines = lines.for_each(|item| {
        println!("> {:?}", item);
        Ok(())
    });

    std::thread::spawn(move || {
        std::thread::sleep(Duration::from_millis(5000));
        println!("system shutting down");
        let _ = tx.send(());
    });

    let lines = lines.select2(rx);

    tokio::run(lines.map(|_| ()).map_err(|_| ()));
}

我怎样才能停止阅读这个？

Answer 1

您的策略没有问题，但它只适用于不通过 Tokio 的 blocking 执行阻塞操作的期货（传统的阻塞类型 should never be done inside a future）。

您可以通过将tokio::io::lines(..) 未来替换为简单的区间未来来测试：

let lines = Interval::new(Instant::now(), Duration::from_secs(1));

---

问题是 tokio::io::Stdin 内部使用 tokio_threadpool::blocking 。

当您使用Tokio thread pool blocking（强调我的）时：

注意：调用阻塞的整个任务都会被阻塞，只要 提供了闭包块，即使你使用了未来的组合器这样 as select - 此任务中的其他期货将不会取得进展 直到关闭返回。如果不希望这样做，请确保 blocking runs in its自己的任务（例如使用 futures::sync::oneshot::spawn)。

由于这将阻塞组合器中的所有其他未来，因此您的 Receiver 将无法从 Sender 获得信号，直到阻塞结束。

请参阅How can I read non-blocking from stdin? 或者您可以使用tokio-stdin-stdout，它创建一个通道来使用来自stdin 线程的数据。它还有一个逐行示例。

Answer 2

感谢您的评论和纠正我的句子。

我试图阻止这个非阻塞Future 并且成功了。

let lines = Interval::new(Instant::now(), Duration::from_secs(1));

我的理解是，在这种情况下，用 tokio threadpool::blocking 包裹阻塞 Future 会起作用。 我稍后再试试。

非常感谢。