0

当我尝试缓存一个值，只要它是有效的，并在它变得无效时更新它，我就会遇到一个问题。我相信这个问题是由于我试图在异步执行之间共享状态。此外，该组件存在于多线程/并发环境中。

我看到的我不知道如何解决的错误是

future is not `Send` as this value is used across an await

以下是我可以提出的一个最小示例（它还具有一些所有权问题），它通常捕获了我的用例和我看到的问题。 这是代码的游乐场。

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use std::sync::{Arc, Mutex};

struct Creds {
    expires_at: DateTime<Utc>,
}

impl Creds {
    fn is_expired(&self) -> bool {
        self.expires_at.le(&Utc::now())
    }
}

#[async_trait]
trait CredsProvider {
    async fn get_creds(&self) -> Creds;
}

struct MyCredsProvider {
    cached_creds: Arc<Mutex<Option<Creds>>>,
}

impl MyCredsProvider {
    fn new() -> Self {
        MyCredsProvider {
            cached_creds: Arc::new(Mutex::new(None)),
        }
    }
    async fn inner_get_creds(&self) -> Creds {
        todo!()
    }
}

#[async_trait]
impl CredsProvider for MyCredsProvider {
    async fn get_creds(&self) -> Creds {
        let mg = self
            .cached_creds
            .lock()
            .expect("Unable to get lock on creds mutex");
        if mg.is_some() && !mg.as_ref().unwrap().is_expired() {
            return mg.unwrap();
        }
        let new_creds = self.inner_get_creds().await;
        *mg = Some(new_creds);
        return new_creds;
    }
}

#[tokio::main]
async fn main() {
    MyCredsProvider::new();
    // Some multi-threaded / concurrent logic to periodically refresh creds
    todo!()
}

我不确定如何在示例中包含此内容，但main可以想象多个工作线程同时/并行运行每个调用CredsProvider.get_creds，然后使用这些凭据执行一些工作（如果您可以将其添加到完整的工作示例中，那就是非常感谢我的熏陶）。假设MyCredsProvider.inner_get_creds是昂贵的，并且应该只在缓存的凭据过期时调用。

我该如何解决这个问题？我以为Arc<Mutex<>>这就足够了，但似乎还不够。有一次，我尝试制作Creds和特质，以便我可以拥有Arc<Mutex<Option<Box<dyn Creds + Send + Sync>>>>，但感觉像是错误的道路并且没有奏效。

谢谢。

4

1 回答 1

0

您可能想切换到tokio::sync::Mutex ( playground )。

它解决了

future is not `Send` as this value is used across an await

代码：

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use std::sync::Arc;
use tokio::sync::Mutex;

#[derive(Clone)]
struct Creds {
    expires_at: DateTime<Utc>,
}

impl Creds {
    fn is_expired(&self) -> bool {
        self.expires_at.le(&Utc::now())
    }
}

#[async_trait]
trait CredsProvider {
    async fn get_creds(&self) -> Creds;
}

struct MyCredsProvider {
    cached_creds: Arc<Mutex<Option<Creds>>>,
}

impl MyCredsProvider {
    fn new() -> Self {
        MyCredsProvider {
            cached_creds: Arc::new(Mutex::new(None)),
        }
    }
    async fn inner_get_creds(&self) -> Creds {
        todo!()
    }
}

#[async_trait]
impl CredsProvider for MyCredsProvider {
    async fn get_creds(&self) -> Creds {
        let mut mg = self
            .cached_creds
            .lock()
            .await;
        if mg.is_some() && !mg.as_ref().unwrap().is_expired() {
            return mg.clone().unwrap();
        } else {
            let new_creds = self.inner_get_creds().await;
            *mg = Some(new_creds.clone());
            return new_creds;
        }
    }
}

#[tokio::main]
async fn main() {
    MyCredsProvider::new();
    // Some multi-threaded / concurrent logic to periodically refresh creds
    todo!()
}
于 2020-11-28T19:53:32.930 回答