我正在尝试创建一个系统，我的应用程序可以通过该系统从 Redis PubSub channel 接收流数据并进行处理。 Redis driver我正在使用的，以及我见过的所有其他 Rust 的 Redis 驱动程序，使用阻塞操作从 channel 获取数据，该 channel 仅在接收数据时返回值:

let msg = match pubsub.get_message() {
        Ok(m) => m,
        Err(_) => panic!("Could not get message from pubsub!")
};
let payload: String = match msg.get_payload() {
    Ok(s) => s,
    Err(_) => panic!("Could not convert redis message to string!")
};

我想使用 futures-rs库在将来包装这个阻塞函数调用，这样我就可以在等待输入的同时在我的应用程序中执行其他任务。

我读了tutorial用于 future ，并尝试创建一个 Stream，当 PubSub 收到数据时会发出信号，但我不知道该怎么做。

如何为阻塞的 pubsub.get_message() 函数创建 schedule 和 poll 函数？

最佳答案

重要警告 我以前从未使用过这个库，而且我对某些概念的低级知识有点......缺乏。大多数情况下，我正在通读 the tutorial .我很确定任何做过异步工作的人都会读到这篇文章并大笑，但这对其他人来说可能是一个有用的起点。买者自负!

让我们从更简单的事情开始，演示 Stream 的工作原理。我们可以将 Result 的迭代器转换为流:

extern crate futures;

use futures::Future;
use futures::stream::{self, Stream};

fn main() {
    let payloads: Vec<Result<String, ()>> = vec![Ok("a".into()), Ok("b".into())];
    let payloads = stream::iter(payloads.into_iter());

    let foo = payloads
        .and_then(|payload| futures::finished(println!("{}", payload)))
        .for_each(|_| Ok(()));

    foo.forget();
}

这向我们展示了一种使用流的方法。我们使用 and_then 对每个有效载荷做一些事情(这里只是打印出来)，然后使用 for_each 将 Stream 转换回 future 。然后我们可以通过调用奇怪的名字来运行 future forget method .

下一步是将 Redis 库绑定(bind)到组合中，只处理一条消息。由于 get_message() 方法是阻塞的，我们需要引入一些线程。在这种类型的异步系统中执行大量工作不是一个好主意，因为其他一切都会被阻塞。 For example :

Unless it is otherwise arranged to be so, it should be ensured that implementations of this function finish very quickly.

在理想的世界中，redis crate 将像 futures 一样构建在库之上，并在本地公开所有这些内容。

extern crate redis;
extern crate futures;

use std::thread;
use futures::Future;
use futures::stream::{self, Stream};

fn main() {
    let client = redis::Client::open("redis://127.0.0.1/").expect("Unable to connect to Redis");

    let mut pubsub = client.get_pubsub().expect("Unable to create pubsub handle");
    pubsub.subscribe("rust").expect("Unable to subscribe to redis channel");

    let (tx, payloads) = stream::channel();

    let redis_thread = thread::spawn(move || {
        let msg = pubsub.get_message().expect("Unable to get message");
        let payload: Result<String, _> = msg.get_payload();
        tx.send(payload).forget();
    });

    let foo = payloads
        .and_then(|payload| futures::finished(println!("{}", payload)))
        .for_each(|_| Ok(()));

    foo.forget();
    redis_thread.join().expect("unable to join to thread");
}

我的理解在这里变得更加模糊。在一个单独的线程中，我们阻塞消息并在我们收到消息时将其推送到 channel 中。我不明白的是为什么我们需要捕获线程的句柄。我希望 foo.forget 会阻塞自己，等待流为空。

在与 Redis 服务器的 telnet 连接中，发送:

publish rust awesome

你会看到它起作用了。添加打印语句显示(对我而言)foo.forget 语句在线程生成之前运行。

多条消息比较棘手。 Sender 消耗自身以防止生成方比消耗方领先太多。这是通过从 send 返回另一个 future 来完成的!我们需要将它从那里穿梭回来，以便在循环的下一次迭代中重用它:

extern crate redis;
extern crate futures;

use std::thread;
use std::sync::mpsc;

use futures::Future;
use futures::stream::{self, Stream};

fn main() {
    let client = redis::Client::open("redis://127.0.0.1/").expect("Unable to connect to Redis");

    let mut pubsub = client.get_pubsub().expect("Unable to create pubsub handle");
    pubsub.subscribe("rust").expect("Unable to subscribe to redis channel");

    let (tx, payloads) = stream::channel();

    let redis_thread = thread::spawn(move || {
        let mut tx = tx;

        while let Ok(msg) = pubsub.get_message() {
            let payload: Result<String, _> = msg.get_payload();

            let (next_tx_tx, next_tx_rx) = mpsc::channel();

            tx.send(payload).and_then(move |new_tx| {
                next_tx_tx.send(new_tx).expect("Unable to send successor channel tx");
                futures::finished(())
            }).forget();

            tx = next_tx_rx.recv().expect("Unable to receive successor channel tx");
        }
    });

    let foo = payloads
        .and_then(|payload| futures::finished(println!("{}", payload)))
        .for_each(|_| Ok(()));

    foo.forget();
    redis_thread.join().expect("unable to join to thread");
}

我相信随着时间的推移，这种类型的互操作将会有更多的生态系统。例如，futures-cpupool crate 可以可能扩展以支持与此类似的用例。

关于asynchronous - 如何使用 futures.rs 和 Redis PubSub 为阻塞调用实现 future 流？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/38906680/