在實(shí)際開發(fā)過程中，我們可能會(huì)遇到并發(fā)寫文件的場(chǎng)景，如果處理不當(dāng)很可能出現(xiàn)文件內(nèi)容亂序問題。下面我們通過一個(gè)示例程序描述這一過程并給出解決該問題的方法。

use std::{
    fs::{self, File, OpenOptions},
    io::{Write},
    sync::Arc,
    time::{SystemTime, UNIX_EPOCH},
};
use tokio::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::new().unwrap();
    rt.block_on(async {
        loop {
            while set.len() >= max_tasks {
                set.join_next().await;
            }
            未做寫互斥函數(shù)
            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("
".as_bytes()).unwrap();
        file.write_all("
".as_bytes()).unwrap();
    }
}

代碼不復(fù)雜，tokio 實(shí)現(xiàn)一個(gè)并發(fā)runtime，寫文件函數(shù)是直接寫時(shí)間戳，為了方便展示亂序所以寫入兩次換行。

輸出的文本大概長(zhǎng)這樣：

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

很明顯，寫入并未達(dá)到預(yù)期，間隔并不平均，函數(shù)內(nèi)部的執(zhí)行步驟是亂序的。

我們把上面的程序改造一下：

use std::{
    fs::{self, File, OpenOptions},
    io::Write,
    sync::Arc,
    time::{SystemTime, UNIX_EPOCH},
};
use tokio::Mutex;
use tokio::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::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>) {
    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("
".as_bytes()).unwrap();
        f.write_all("
".as_bytes()).unwrap();
    }
}

這次我們用到了tokio::Mutex，write_line_mutex函數(shù)在每次執(zhí)行寫任務(wù)以前先獲取文件互斥鎖。

看看這次的文件內(nèi)容：

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

寫入的格式正確，保證每次函數(shù)寫函數(shù)完整執(zhí)行。

關(guān)于文件寫互斥這點(diǎn)事兒，今兒就聊到這。

審核編輯：劉清