在异步 rust 中处理重复插入数据库答案

【问题标题】：Handling duplicate inserts into database in async rust在异步 rust 中处理重复插入数据库
【发布时间】：2021-08-24 20:19:28
【问题描述】：

这里是 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 作为我的网络服务器/数据库库。

【问题讨论】：

听起来您应该更改 SQL 逻辑以使用 ON CONFLICT。我认为这是一种反模式，可以检查一个查询是否存在某些内容，然后将其插入另一个查询，这正是您遇到的原因。
这看起来像是“我宁愿一开始就没有这个问题”的情况。为什么任务 1 和任务 2 指向同一条推文？
“从任务 1 读取 > 从任务 2 读取 > 从任务 1 写入（成功）> 从任务 2 写入（错误）。”挑剔：INSERT，具体来说。使用UPDATE，您可以在读取时锁定行，确保第二个任务无法访问。
哇，每天都能学到新东西。我不知道ON CONFLICT。我已经重写了我的代码，现在它既正确又更快。非常感谢@loganfsmyth。想发布一个答案，我会标记它是正确的？

标签： rust actix-web rust-actix rust-sqlx

【解决方案1】：

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

if (!exists) {
  writeValue()
}

需要被某种锁保护，或者需要更改代码以自动写入，因为写入可能会失败，因为已经写入了其他内容。

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

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

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

【讨论】：