Rust多线程异步任务借用数据逃逸问题及解决方案探讨

错误详情 #

error[E0521]: borrowed data escapes outside of method
  --> client_runner/src/lib.rs:37:23
   |
15 |     pub async fn run(&mut self) {
   |                      ---------
   |                      |
   |                      `self` is a reference that is only valid in the method body
   |                      let's call the lifetime of this reference `'1`
...
37 |         for client in self.clients.iter() {
   |                       ^^^^^^^^^^^^^^^^^^^
   |                       |
   |                       `self` escapes the method body here
   |                       argument requires that `'1` must outlive `'static`

问题背景与困惑 #

我正在开发一个项目，需要在独立线程中从两个数据源流式获取数据。我理解这个错误是因为clients是self的成员，在run方法中被借用，但不确定如何修正。

我尝试过通过索引遍历向量，用self.clients[idx].stream()访问客户端，但启动Tokio任务时仍然失败；如果不启动Tokio任务，客户端能正常运行但会变成同步执行。

我有Java、C++和Python的OOP开发背景，刚接触Rust，希望能得到适配Rust所有权模型的“Rust风格”解决方案，也欢迎指出我的错误并提供改进建议。

相关代码 #

crate main_project #

main.rs

use client_runner::*;
use data_sources::{
    DataSource,
    source1_interface::Source1Client,
    source2_interface::Source2Client,
};

#[tokio::main(flavor = "multi_thread")]
async fn main() {
    // 创建客户端向量
    let client_list: Vec<Box<dyn DataSource + Send + Sync>> = vec! [
        Box::new(Source1Client::new()),
        Box::new(Source2Client::new()),
    ];
    
    // 传入客户端向量，在独立线程中启动
    let mut client_runner = ClientRunner::new(client_list);
    client_runner.run().await;
}

crate client_runner #

lib.rs

use data_sources::DataSource;
use std::thread;

pub struct ClientRunner {
    clients: Vec<Box<dyn DataSource + Send + Sync>>,
}

impl ClientRunner {
    /// 创建新的ClientRunner，参数是实现了DataSource trait的客户端列表
    pub fn new(client_list: Vec<Box<dyn DataSource + Send + Sync>>) -> Self {
        ClientRunner {clients: client_list}
    }

    pub async fn run(&mut self) {
        println!("主线程ID: {:?}", thread::current().id());
        // 单次调用可以运行但不是多线程
        // self.clients[0].stream().await;
        
        // 此处报错
        for client in self.clients.iter() {
            // 启动工作线程：
            // 流式获取数据
            tokio::task::spawn(async move {
                client.stream().await;
            });
        }
    }
}

crate data_sources #

lib.rs

use async_trait::async_trait;

#[async_trait]
pub trait DataSource {
    async fn stream(&self);
}

pub mod source1_interface;
pub mod source2_interface;

source1_interface.rs

pub struct Source1Client{
    ...
}

unsafe impl Send for Source1Client {}
unsafe impl Sync for Source1Client {}

#[async_trait]
impl DataSource for Source1Client {
    async fn stream(&self) {
        println!("Source1 流式获取数据，线程ID: {:?}", thread::current().id());
    }
}

impl Source1Client {
    pub fn new() -> Self {
        ...
        Source1Client {
            ...
        }
    }

    ...
}

source2_interface.rs

pub struct Source2Client{
    ...
}

unsafe impl Send for Source2Client {}
unsafe impl Sync for Source2Client {}

#[async_trait]
impl DataSource for Source2Client {
    async fn stream(&self) {
        println!("Source2 流式获取数据，线程ID: {:?}", thread::current().id());
    }
}

impl Source2Client {
    pub fn new() -> Self {
        ...
        Source2Client {
            ...
        }
    }

    ...
}

已尝试的解决方案 #

我发现可以用Arc智能指针替代Box。Arc<T>提供对堆上分配的T类型值的共享所有权，调用Arc::clone会生成新的Arc实例，指向同一堆分配并增加引用计数，这样就解决了借用数据逃逸的问题。

修改后的代码如下：

crate main_project #

main.rs

...

#[tokio::main(flavor = "multi_thread")]
async fn main() {
    // 创建客户端向量
    let client_list: Vec<Arc<dyn DataSource + Send + Sync>> = vec! [
        Arc::new(Source1Client::new()),
        Arc::new(Source2Client::new()),
    ];
    
    ...
}

crate client_runner #

lib.rs

...
use std::sync::Arc;

pub struct ClientRunner {
    clients: Vec<Arc<dyn DataSource + Send + Sync>>,
}

impl ClientRunner {
    pub fn new(client_list: Vec<Arc<dyn DataSource + Send + Sync>>) -> Self {
        ClientRunner { clients: client_list }
    }

    pub async fn run(&mut self) {
        println!("主线程ID: {:?}", thread::current().id());
        
        let mut worker_tasks = Vec::new();
        for idx in 0..self.clients.len() {
            // 启动工作任务：
            // 流式获取数据
            let moved_client = Arc::clone(&self.clients[idx]);
            let task_handle = tokio::task::spawn(async move {
                moved_client.stream().await;
            });
            worker_tasks.push(task_handle);
        }
 
        // 等待所有任务完成
        for task in worker_tasks {
            tokio::join!(task);
        }
    }
}

希望能得到更贴合Rust风格的解决方案。

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