文盘Rust -- Mutex解决并发写文件乱序问题

rust,mutex,解决,并发,文件,乱序,问题 · 浏览次数 : 39

小编点评

**解决并发写文件问题的方法：** 1. **使用同步锁：**在写操作之前获取文件互斥锁，确保只有一个线程可以写文件。 2. **使用 `Arc` 或 `Mutex`：**创建一个共享变量，并在写操作之前将其锁定，确保多个线程可以访问该变量。 3. **使用 `tokio::sync::Mutex`：**创建一个单独的互斥锁，并在每个写操作之前获取该锁。 4. **使用异步操作：**将写操作封装为异步操作，并使用 `tokio::spawn` 创建多个线程执行该操作。 5. **使用 `Semaphore`：**创建一个信号量，并在写操作之前将其加到信号量中。 6. **使用 `RwLock`：**使用 `RwLock` 类来访问文件，它可以确保多个线程可以读取或写入文件，并在写操作之前将其加到信号量中。 **示例代码：** ```rust use std::fs::{self, File, OpenOptions}; use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; fn main() { println!(\"parallel write file!\"); let max_tasks = 200; let file_ref = OpenOptions::new() .create(true) .write(true) .append(true) .open("/tmp/parallel") .unwrap(); let f = Arc::new(Mutex::new(file_ref)); let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async { loop { while f.lock().await.len() >= max_tasks { f.lock().await.join_next().await; } let mut file = Arc::clone(&f.lock().await); for i in 0..1000 { let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap(); let content = now.as_secs().to_string(); file.write_all(content.as_bytes()).unwrap(); file.write_all("\n".as_bytes()).unwrap(); } } }); } ``` **注意：** * 在使用任何锁或互斥锁之前，请确保文件已打开。 * 使用 `Arc` 或 `Mutex` 时，请注意线程安全问题。 * 使用 `tokio::sync::Mutex` 或 `RwLock` 时，请确保所有线程都使用相同的锁。

正文

在实际开发过程中，我们可能会遇到并发写文件的场景，如果处理不当很可能出现文件内容乱序问题。下面我们通过一个示例程序描述这一过程并给出解决该问题的方法。

use std::{
    fs::{self, File, OpenOptions},
    io::{Write},
    sync::Arc,
    time::{SystemTime, UNIX_EPOCH},
};
use tokio::task::JoinSet;

fn main() {
    println!("parallel write file!");
    let max_tasks = 200;
    let _ = fs::remove_file("/tmp/parallel");
    let file_ref = OpenOptions::new()
        .create(true)
        .write(true)
        .append(true)
        .open("/tmp/parallel")
        .unwrap();

    let mut set: JoinSet<()> = JoinSet::new();
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(async {
        loop {
            while set.len() >= max_tasks {
                set.join_next().await;
            }
            未做写互斥函数
            let mut file_ref = OpenOptions::new()
                .create(true)
                .write(true)
                .append(true)
                .open("/tmp/parallel")
                .unwrap();
            set.spawn(async move { write_line(&mut file_ref) });
        }
    });
}

fn write_line(file: &mut File) {
    for i in 0..1000 {
        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
        let mut content = now.as_secs().to_string();
        content.push_str("_");
        content.push_str(&i.to_string());

        file.write_all(content.as_bytes()).unwrap();
        file.write_all("\n".as_bytes()).unwrap();
        file.write_all("\n".as_bytes()).unwrap();
    }
}

代码不复杂，tokio 实现一个并发runtime，写文件函数是直接写时间戳，为了方便展示乱序所以写入两次换行。

输出的文本大概长这样

1691287258_979





1691287258_7931691287258_301

1691287258_7431691287258_603

1691287258_8941691287258_47






1691287258_895
1691287258_553

1691287258_950
1691287258_980


1691287258_48
1691287258_302

1691287258_896
1691287258_744




1691287258_6041691287258_554

很明显，写入并未达到预期，间隔并不平均，函数内部的执行步骤是乱序的。

我们把上面的程序改造一下

use std::{
    fs::{self, File, OpenOptions},
    io::Write,
    sync::Arc,
    time::{SystemTime, UNIX_EPOCH},
};
use tokio::sync::Mutex;
use tokio::task::JoinSet;

fn main() {
    println!("parallel write file!");
    let max_tasks = 200;
    let _ = fs::remove_file("/tmp/parallel");
    let file_ref = OpenOptions::new()
        .create(true)
        .write(true)
        .append(true)
        .open("/tmp/parallel")
        .unwrap();

    let f = Arc::new(Mutex::new(file_ref));

    let mut set: JoinSet<()> = JoinSet::new();
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(async {
        loop {
            while set.len() >= max_tasks {
                set.join_next().await;
            }

            let mut file = Arc::clone(&f);
            set.spawn(async move { write_line_mutex(&mut file).await });
        }
    });
}

async fn write_line_mutex(mutex_file: &Arc<Mutex<File>>) {
    for i in 0..1000 {
        let mut f = mutex_file.lock().await;
        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
        let mut content = now.as_secs().to_string();
        content.push_str("_");
        content.push_str(&i.to_string());

        f.write_all(content.as_bytes()).unwrap();
        f.write_all("\n".as_bytes()).unwrap();
        f.write_all("\n".as_bytes()).unwrap();
    }
}

这次我们用到了tokio::sync::Mutex，write_line_mutex函数在每次执行写任务以前先获取文件互斥锁。

看看这次的文件内容

1691288040_374

1691288040_374

1691288040_374

1691288040_375

1691288040_374

1691288040_374

1691288040_374

1691288040_374

1691288040_374

1691288040_374

1691288040_374

1691288040_374

1691288040_374

1691288040_374

1691288040_375

1691288040_375

1691288040_374

1691288040_375

1691288040_375

1691288040_375

1691288040_375

1691288040_375

1691288040_375

1691288040_375

1691288040_375

1691288040_375

1691288040_375

写入的格式正确，保证每次函数写函数完整执行。

关于文件写互斥这点事儿，今儿就聊到这。

完整源码

作者：京东科技贾世闻

来源：京东云开发者社区