在 async rust 中处理重复插入到数据库中

问题描述

这里是 Rust 和异步编程的初学者。

我有一个函数可以下载并在数据库中存储一堆推文：

pub async fn process_user_timeline(config: &Settings,pool: &PgPool,user_object: &Value) {
    // get timeline
    if let Ok((user_timeline,_)) =
        get_user_timeline(config,user_object["id"].as_str().unwrap()).await
    {
        // store tweets
        if let Some(tweets) = user_timeline["data"].as_array() {
            for tweet in tweets.iter() {
                store_tweet(pool,&tweet,&user_timeline,"normal")
                    .await
                    .unwrap_or_else(|e| {
                        println!(
                            ">>>X>>> Failed to store tweet {}: {:?}",tweet["id"].as_str().unwrap(),e
                        )
                    });
            }
        }
    }
}

它被另一个函数在异步循环中调用：

pub async fn loop_until_hit_rate_limit<'a,T,Fut>(
    object_arr: &'a [T],settings: &'a Settings,pool: &'a PgPool,f: impl Fn(&'a Settings,&'a PgPool,&'a T) -> Fut + copy,rate_limit: usize,) where
    Fut: Future,{
    let total = object_arr.len();
    let capped_total = min(total,rate_limit);

    let mut futs = vec![];
    for (i,object) in object_arr[..capped_total].iter().enumerate() {
        futs.push(async move {
            println!(">>> PROCESSING {}/{}",i + 1,total);
            f(settings,pool,object).await;
        });
    }
    futures::future::join_all(futs).await;
}

有时两个异步任务会尝试同时插入相同的推文，从而产生此错误：

Failed to store tweet 1398307091442409475: Database(PgDatabaseError { severity: Error,code: "23505",message: "duplicate key value violates unique constraint \"tweets_tweet_id_key\"",detail: Some("Key (tweet_id)=(1398307091442409475) already exists."),hint: None,position: None,where: None,schema: Some("public"),table: Some("tweets"),column: None,data_type: None,constraint: Some("tweets_tweet_id_key"),file: Some("nbtinsert.c"),line: Some(656),routine: Some("_bt_check_unique") })

请注意，代码在插入推文之前已经检查是否存在推文，因此这只发生在以下情况：从任务 1 读取 > 从任务 2 读取 > 从任务 1 写入（成功）> 从任务 2 写入（错误）。

为了解决这个问题，我迄今为止最好的尝试是放置一个 unwrap_or_else() 子句，它可以让其中一个任务失败而不会在整个执行过程中出现恐慌。我知道至少有一个缺点 - 有时两项任务都会完成，推文永远不会被写出来。它发生在

我不知道我的方法还有其他缺点吗？

处理这个问题的正确方法是什么？我讨厌丢失数据，更糟糕的是不确定性地丢失数据。

PS 我使用 actix web 和 sqlx 作为我的网络服务器/数据库库。

解决方法

通常对于可能由多个线程/进程编写的任何内容，任何逻辑，如

if (!exists) {
  writeValue()
}

要么需要某种锁的保护，要么需要更改代码以进行原子写入，但写入可能会失败，因为其他内容已经写入。

对于 Rust 中的内存数据，您可以使用 Mutex 来确保您可以读取数据，然后在其他任何读取数据之前将其写回，或者使用 Atomic 以这样的方式修改数据方式，如果已经写了它，你可以检测到。

在数据库中，对于可能与同时发生的其他查询发生冲突的任何查询，您需要在查询中使用 ON CONFLICT 子句，以便数据库本身知道在尝试时要做什么写入数据，它已经存在。

对于你的情况，因为我猜推文是不可变的，你可能想要做 ON CONFLICT tweet_id DO NOTHING（或任何你的 ID 列），在这种情况下，INSERT 将跳过插入，如果已经有一条带有您插入的 ID 的推文，它不会抛出错误。