Rust中B树异步递归遍历函数触发‘无法解析不透明类型’错误

我需要实现一个对B树元素进行中序遍历的函数，由于访问子节点是异步操作，函数要返回AsyncStream。我用async_stream宏创建流，搭配tokio和tokio_stream库使用相关trait，但递归实现时编译器报错。

最小复现代码 #

use async_stream::stream;
use std::sync::Arc;
use tokio::sync::RwLock;
use tokio_stream::{Stream, StreamExt};

struct Node {
    keys: Vec<usize>,
    children: Vec<Arc<RwLock<Node>>>,
}

impl Node {
    fn new(keys: Vec<usize>, children: Vec<Arc<RwLock<Node>>>) -> Arc<RwLock<Self>> {
        Arc::new(RwLock::new(Node { keys, children }))
    }
    async fn get_child(&self, index: usize) -> Option<Arc<RwLock<Node>>> {
        self.children.get(index).cloned()
    }
}

fn traverse(node: Arc<RwLock<Node>>) -> impl Stream<Item = usize> {
    stream! {
        let n_keys = node.read().await.keys.len();
        for i in 0..n_keys {
            yield node.read().await.keys[i].clone();
            if let Some(child) = node.read().await.get_child(i).await {
                let child_stream = traverse(child);
                tokio::pin!(child_stream);
                while let Some(k) = child_stream.next().await {
                    yield k
                }
            }
        }
        // last child
        if let Some(child) = node.read().await.get_child(n_keys).await {
            let child_stream = traverse(child);
            tokio::pin!(child_stream);
            while let Some(k) = child_stream.next().await {
                yield k
            }
        }

    }
}

#[tokio::test]
async fn t() -> Result<(), Box<dyn std::error::Error>> {
    let root = Node::new(
        vec![3],
        vec![Node::new(vec![1], vec![]), Node::new(vec![5], vec![])],
    );
    let mut result = vec![];
    let stream = traverse(root);
    tokio::pin!(stream);
    while let Some(k) = stream.next().await {
        result.push(k);
    }
    assert_eq!(result, vec![1, 3, 5]);
    Ok(())
}

报错信息 #

error[E0720]: cannot resolve opaque type
   --> tests/streamq.rs:20:41
    |
15  |       async fn get_child(&self, index: usize) -> Option<Arc<RwLock<Node>>> {
    |                                                  ------------------------- returning this type `AsyncStream<usize, [async block@/home/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`
...
20  |   fn traverse(node: Arc<RwLock<Node>>) -> impl Stream<Item = usize> {
    |                                           ^^^^^^^^^^^^^^^^^^^^^^^^^ recursive opaque type
21  | /     stream! {
22  | |         let n_keys = node.read().await.keys.len();
23  | |         for i in 0..n_keys {
24  | |             yield node.read().await.keys[i].clone();
25  | |             if let Some(child) = node.read().await.get_child(i).await {
26  | |                 let child_stream = traverse(child);
    | |                     ------------ async closure captures itself here
27  | |                 tokio::pin!(child_stream);
    | |                 ------------------------- async closure captures itself here
28  | |                 while let Some(k) = child_stream.next().await {
    | |                                     -------------------
    | |                                     |
    | |                                     async closure captures itself here
    | |                                     async closure captures itself here
...   |
41  | |
42  | |     }
    | |     -
    | |_____|
    | |_____returning here with type `AsyncStream<usize, [async block@/home/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`
    | |_____returning here with type `AsyncStream<usize, [async block@/home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`
    | |_____returning here with type `AsyncStream<usize, [async block@/home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`
    | |_____returning here with type `AsyncStream<usize, [async block@/home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`
    | |_____returning here with type `AsyncStream<usize, [async block@/home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`
    | |_____returning here with type `AsyncStream<usize, [async block@/home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`
    |       returning here with type `AsyncStream<usize, [async block@/home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`
    |
   ::: /home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.34.0/src/sync/rwlock.rs:430:33
    |
430 |       pub async fn read(&self) -> RwLockReadGuard<'_, T> {
    |                                   ---------------------- returning this type `AsyncStream<usize, [async block@/home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`
    |
   ::: /home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/yielder.rs:42:41
    |
42  |       pub fn send(&mut self, value: T) -> impl Future<Output = ()> {
    |                                           ------------------------- returning this type `AsyncStream<usize, [async block@/home/g.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59]>`

For more information about this error, try `rustc --explain E0720`.
error: could not compile `` (test "streamq") due to previous error
warning: build failed, waiting for other jobs to finish...

我理解问题出在-> impl Stream<...>是不透明类型，递归调用时编译器无法推断其具体类型，导致报错。有没有无需手动实现Stream trait的解决方法？

方法一：使用BoxStream装箱递归流 #

把返回类型改为BoxStream<'static, usize>，这样所有递归产生的流都会被装箱成统一的 trait object，编译器不需要推断具体的递归类型。

修改后的traverse函数：

use tokio_stream::StreamExt;
type BoxStream<T> = std::pin::Pin<Box<dyn tokio_stream::Stream<Item = T> + Send + 'static>>;

fn traverse(node: Arc<RwLock<Node>>) -> BoxStream<usize> {
    Box::pin(stream! {
        let node_guard = node.read().await;
        let n_keys = node_guard.keys.len();
        drop(node_guard); // 提前释放锁，避免持有过久

        for i in 0..n_keys {
            // 重新获取锁读取当前key
            let key = node.read().await.keys[i];
            yield key;
            
            if let Some(child) = node.read().await.get_child(i).await {
                let child_stream = traverse(child);
                tokio::pin!(child_stream);
                while let Some(k) = child_stream.next().await {
                    yield k;
                }
            }
        }

        // 处理最后一个子节点
        if let Some(child) = node.read().await.get_child(n_keys).await {
            let child_stream = traverse(child);
            tokio::pin!(child_stream);
            while let Some(k) = child_stream.next().await {
                yield k;
            }
        }
    })
}

这里还优化了锁的使用：提前释放锁，避免在循环中长时间持有读锁，提升并发性能。

方法二：用非递归方式模拟中序遍历 #

用栈保存待处理的节点和遍历状态，避免递归带来的类型问题。这种方式不需要递归调用traverse，编译器可以正常推断返回的Stream类型。

示例实现：

fn traverse(node: Arc<RwLock<Node>>) -> impl Stream<Item = usize> {
    stream! {
        // 栈元素：(节点, 当前处理到的key索引)
        let mut stack = vec![(node, 0)];

        while let Some((node, mut idx)) = stack.pop() {
            let node_guard = node.read().await;
            let n_keys = node_guard.keys.len();

            if idx < n_keys {
                // 先把当前节点和下一个索引压回栈，处理右子节点前先处理当前节点和左子节点
                stack.push((node.clone(), idx + 1));

                // 压入右子节点（如果存在），后续处理
                if let Some(child) = node_guard.get_child(idx + 1).await {
                    stack.push((child, 0));
                }

                // 输出当前key
                yield node_guard.keys[idx];

                // 压入左子节点（如果存在），下一次循环处理
                if let Some(child) = node_guard.get_child(idx).await {
                    stack.push((child, 0));
                }
            } else if idx == n_keys {
                // 处理最后一个子节点
                if let Some(child) = node_guard.get_child(n_keys).await {
                    stack.push((child, 0));
                }
            }
        }
    }
}

内容的提问来源于stack exchange，提问作者cowlicks