如何创建重复调用异步方法的Tokio Stream？生命周期问题求解

先来看你的代码和遇到的问题：

原始代码 #

use std::future::Future;
use std::pin::Pin;
use std::task::{Poll, Context};
use futures::ready;
use hyper::{
    client::{Client, HttpConnector},
    body::Body,
};
use pin_project::pin_project;
use tokio::stream::{Stream, StreamExt};

// Placeholder for actual result type
#[derive(Debug)]
struct MyResult;

#[pin_project]
struct ResultProducer {
    client: Client<HttpConnector, Body>,
    #[pin]
    current_request: Option<Pin<Box<dyn Future<Output=MyResult>>>>,
}

impl ResultProducer {
    fn new() -> ResultProducer {
        ResultProducer {
            client: Client::new(),
            current_request: None,
        }
    }

    // This is the method that should be called repeatedly
    async fn next_result(&self) -> MyResult {
        let url = "http://www.example.com".parse().unwrap();
        let _ = self.client.get(url).await.unwrap();
        MyResult
    }
}

impl Stream for ResultProducer {
    type Item = MyResult;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let mut fut = self.project().current_request;
        if fut.is_none() {
            fut.set(Some(Box::pin(self.next_result())));
        }

        let res = ready!(fut.as_mut().as_pin_mut().unwrap().poll(cx));
        fut.set(None);
        Poll::Ready(Some(res))
    }
}

#[tokio::main]
async fn main() {
    let requester = ResultProducer::new();
    while let Some(res) = requester.next().await {
        println!("Got a result: {:?}", res);
    }
}

编译错误 #

error[E0495]: cannot infer an appropriate lifetime for lifetime parameter in function call due to conflicting requirements
--> src/main.rs:46:40
|
46 | fut.set(Some(Box::pin(self.next_result())));
| ^^^^^^^^^^^

问题根源分析 #

你遇到的核心问题是异步方法捕获的引用生命周期与Stream的poll_next机制不兼容：

1. next_result(&self)是一个接收&self的异步方法，它返回的Future会捕获这个&self引用，这个Future的生命周期完全绑定到&self的生命周期上。
2. 当你把这个Future放到current_request字段中时，Box<dyn Future<Output=MyResult>>默认要求Future是'static生命周期的（因为没有显式标注生命周期参数）。
3. 但poll_next方法中的self是Pin<&mut Self>类型，这个引用的生命周期只在poll_next的单次调用期间有效。如果Future被挂起（比如等待HTTP请求完成），下次调用poll_next时，原来的&self引用可能已经失效了——编译器无法保证这个引用在Future的整个生命周期内都有效，所以抛出了生命周期冲突错误。

解决方案：用Arc<Mutex>分离状态 #

你提到的把状态提取到独立结构体并用Arc<Mutex>包裹的方案是完全可行的，这也是处理这类异步共享状态问题的标准做法之一。核心思路是让异步方法不再捕获&self引用，而是捕获一个可以安全共享的Arc指针，从而摆脱生命周期的绑定。

修正后的代码示例 #

use std::future::Future;
use std::pin::Pin;
use std::task::{Poll, Context};
use std::sync::{Arc, Mutex};
use futures::ready;
use hyper::{
    client::{Client, HttpConnector},
    body::Body,
};
use pin_project::pin_project;
use tokio::stream::{Stream, StreamExt};

#[derive(Debug)]
struct MyResult;

// 将原结构体中的状态提取到独立结构体中
struct ProducerState {
    client: Client<HttpConnector, Body>,
    // 可以在这里添加其他需要可变访问的业务状态
}

#[pin_project]
struct ResultProducer {
    // 用Arc<Mutex>包裹状态，实现安全的共享与可变访问
    state: Arc<Mutex<ProducerState>>,
    #[pin]
    current_request: Option<Pin<Box<dyn Future<Output=MyResult>>>>,
}

impl ResultProducer {
    fn new() -> ResultProducer {
        ResultProducer {
            state: Arc::new(Mutex::new(ProducerState {
                client: Client::new(),
            })),
            current_request: None,
        }
    }

    // 异步方法不再依赖&self，而是接收Arc<Mutex<ProducerState>>的克隆
    async fn next_result(state: Arc<Mutex<ProducerState>>) -> MyResult {
        let url = "http://www.example.com".parse().unwrap();
        // 锁定状态以访问client（如果只需要读操作，用RwLock会更高效）
        let state_guard = state.lock().unwrap();
        let _ = state_guard.client.get(url).await.unwrap();
        MyResult
    }
}

impl Stream for ResultProducer {
    type Item = MyResult;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let mut this = self.project();
        if this.current_request.is_none() {
            // 克隆Arc传递给异步方法，确保Future持有有效的状态引用
            let state_clone = this.state.clone();
            this.current_request.set(Some(Box::pin(Self::next_result(state_clone))));
        }

        let res = ready!(this.current_request.as_mut().as_pin_mut().unwrap().poll(cx));
        this.current_request.set(None);
        Poll::Ready(Some(res))
    }
}

#[tokio::main]
async fn main() {
    // Stream的next()方法需要&mut self，所以这里要声明为mut
    let mut requester = ResultProducer::new();
    while let Some(res) = requester.next().await {
        println!("Got a result: {:?}", res);
    }
}

关键改进点 #

1. 状态分离：把原来属于ResultProducer的client（以及其他业务状态）转移到ProducerState中，用Arc<Mutex>包裹，实现线程安全的共享访问。
2. 摆脱&self依赖：next_result方法现在接收Arc<Mutex<ProducerState>>的克隆，返回的Future捕获的是这个Arc指针，而不是&self引用。Arc的生命周期是'static（只要内部的ProducerState满足），所以可以安全地放入Box<dyn Future>中。
3. 安全的状态访问：通过Mutex锁定状态，确保在异步操作中对状态的访问是线程安全的（如果你的场景中只有读操作，可以换成RwLock来提升并发性能）。

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