Общее состояние в Rust: RwLock, атомики и scoped-потоки

На странице Многопоточность ты делил счётчик между потоками через Arc<Mutex<T>>: Arc позволяет многим потокам владеть значением, а Mutex пускает к нему только один поток за раз. На странице Каналы ты вообще обходился без общих данных и отправлял сообщения. Оба инструмента отличные, и на них можно построить почти что угодно.

Но Mutex похож на единственный ключ от туалета в офисе. Он работает, но это перебор, если ты просто хочешь прочитать объявление на доске в коридоре. В стандартной библиотеке Rust есть ещё несколько инструментов, и каждый лучше подходит для своей работы. На этой странице ты познакомишься с четырьмя: scoped-потоками, RwLock, атомиками и OnceLock/LazyLock. А ещё узнаешь, как избежать единственной ошибки, которую компилятор за тебя не поймает: взаимной блокировки (deadlock).

Scoped-потоки: заимствуем вместо перемещения

С thread::spawn обычно нужен move, а часто и Arc. Дело в том, что запущенный поток может продолжать работать после того, как main ушла дальше, поэтому Rust не даёт ему заимствовать твои локальные переменные.

std::thread::scope решает это одним обещанием: каждый поток, запущенный внутри области (scope), заканчивается до того, как область закончится. Раз потоки не могут пережить твои данные, им разрешено их просто заимствовать. Без Arc, без move, без клонирования.

use std::thread;

fn main() {
    let shop = "Магазин у дома";
    let sales = vec![12, 7, 30, 5, 18, 22, 9, 14];
    let (morning, afternoon) = sales.split_at(4);

    let (am, pm) = thread::scope(|s| {
        let am = s.spawn(|| morning.iter().sum::<i32>());
        let pm = s.spawn(|| afternoon.iter().sum::<i32>());
        (am.join().unwrap(), pm.join().unwrap())
    });

    println!("{shop}: утро {am}, день {pm}, всего за день {}", am + pm);
    println!("всё ещё моё: {:?}", sales);
}

Вывод:

Магазин у дома: утро 54, день 63, всего за день 117
всё ещё моё: [12, 7, 30, 5, 18, 22, 9, 14]

▶ Попробуй сам

Что здесь происходит:

thread::scope(|s| { ... }) открывает область. Потоки запускаются через s.spawn(...) вместо thread::spawn(...).
Два замыкания заимствуют morning и afternoon — это просто срезы вектора sales. Ничего не копируется.
То, что возвращает замыкание области (здесь кортеж из двух сумм), становится значением thread::scope.
После области sales по-прежнему принадлежит main, так что им можно пользоваться дальше.

Даже если ты забудешь вызвать join(), область дождётся всех своих потоков на закрывающей }. Именно это обещание и делает заимствование безопасным.

`RwLock`: много читателей или один писатель

Представь доску с меню в кафе. Её могут одновременно читать много посетителей, и это нормально. Но когда хозяин меняет цену, всем приходится немного подождать, пока не повесят новую доску.

Это и есть RwLock («read-write lock», блокировка чтения-записи). В неё два входа:

read() даёт общую охрану для чтения (read guard). Её могут держать много потоков одновременно.
write() даёт эксклюзивную охрану для записи (write guard). Пока писатель её держит, никто другой не может ни читать, ни писать.

Mutex такой разницы не делает: даже двум читателям приходится ждать друг друга. Вот кеш настроек, который читают несколько обработчиков запросов и один раз обновляет администратор:

use std::sync::{Arc, RwLock};
use std::thread;

struct Settings {
    theme: String,
    max_upload_mb: u32,
}

fn main() {
    let settings = Arc::new(RwLock::new(Settings {
        theme: String::from("light"),
        max_upload_mb: 10,
    }));

    // Четыре обработчика запросов читают настройки одновременно.
    let mut handles = Vec::new();
    for id in 1..=4 {
        let settings = Arc::clone(&settings);
        handles.push(thread::spawn(move || {
            let s = settings.read().unwrap(); // общий доступ: другие тоже могут читать
            format!("запрос {id}: тема={}, лимит={} МБ", s.theme, s.max_upload_mb)
        }));
    }
    for handle in handles {
        println!("{}", handle.join().unwrap());
    }

    // Администратор один раз меняет настройки.
    {
        let mut s = settings.write().unwrap(); // эксклюзивно: все остальные ждут
        s.theme = String::from("dark");
        s.max_upload_mb = 25;
    } // охрана записи удаляется здесь, и блокировка снова свободна

    let s = settings.read().unwrap();
    println!("после обновления: тема={}, лимит={} МБ", s.theme, s.max_upload_mb);
}

Вывод:

запрос 1: тема=light, лимит=10 МБ
запрос 2: тема=light, лимит=10 МБ
запрос 3: тема=light, лимит=10 МБ
запрос 4: тема=light, лимит=10 МБ
после обновления: тема=dark, лимит=25 МБ

▶ Попробуй сам

Потоки могут заканчивать в любом порядке, но вывод всегда одинаковый. Каждый поток возвращает свою строку, а не печатает её, и main выводит строки в том порядке, в каком дожидается потоков.

Когда RwLock лучше Mutex? Когда чтений много, записей мало, и каждое чтение занимает какое-то время (например, поиск по большому кешу). Тогда читатели действительно работают бок о бок.

Когда нет? Если чтения крошечные или записи происходят постоянно, RwLock — это просто более сложный Mutex. Он ведёт дополнительный учёт, а читатели и писатели всё равно в основном ждут друг друга. Если не уверен, начни с Mutex. Он проще, а перейти можно и потом, если измерения покажут, что блокировка — узкое место.

Атомики: счётчики без блокировки

Для одного числа или одного флага «да/нет» даже блокировка — это больше, чем нужно. Атомарные типы, такие как AtomicUsize, AtomicI64 и AtomicBool, могут безопасно меняться многими потоками одновременно, вообще без блокировки. Процессор выполняет каждую операцию за один неделимый шаг, поэтому ни одно обновление не теряется.

Здесь 8 потоков записывают по 1000 просмотров страницы. Ещё они считают голоса: один голос «за», потом три «против», и так снова и снова. Сумма голосов уходит в минус, поэтому это AtomicI64, а не беззнаковый тип:

use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering};
use std::thread;

fn main() {
    let page_views = AtomicUsize::new(0);
    let net_votes = AtomicI64::new(0);

    thread::scope(|s| {
        for _ in 0..8 {
            s.spawn(|| {
                for n in 0..1000 {
                    page_views.fetch_add(1, Ordering::Relaxed);
                    if n % 4 == 0 {
                        net_votes.fetch_add(1, Ordering::Relaxed); // голос «за»
                    } else {
                        net_votes.fetch_sub(1, Ordering::Relaxed); // голос «против»
                    }
                }
            });
        }
    }); // здесь все 8 потоков уже закончили

    println!("просмотры страницы: {}", page_views.load(Ordering::Relaxed));
    println!("итог голосования: {}", net_votes.load(Ordering::Relaxed));
}

Вывод:

просмотры страницы: 8000
итог голосования: -4000

▶ Попробуй сам

Заметь, что page_views не mut. Атомики меняются через общую ссылку — именно поэтому ими могут пользоваться многие потоки одновременно. Методы, которые тебе понадобятся чаще всего:

fetch_add(n, ordering) / fetch_sub(n, ordering) прибавляют или вычитают и возвращают старое значение.
load(ordering) читает текущее значение.
store(value, ordering) перезаписывает его.

Что такое `Ordering`?

Каждая атомарная операция требует Ordering (упорядочивание памяти). Оно говорит процессору и компилятору, насколько строго эта операция должна сохранять порядок относительно окружающих операций с памятью. Полная история глубокая, но новичку хватит двух правил:

Если не уверен, используй Ordering::SeqCst. Это самый строгий и самый безопасный вариант.
Ordering::Relaxed подходит для простого счётчика, который ты читаешь только в конце, как в примере выше. Итог всё равно получается точным.

Если однажды захочешь копнуть глубже, бесплатная книга Rust Atomics and Locks (ссылка внизу страницы) понятно объясняет упорядочивания.

Флаг остановки

AtomicBool — отличный сигнал «пожалуйста, остановись». Здесь фоновый воркер продолжает автосохранение, пока main не переключит флаг:

use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::Duration;

fn main() {
    let stop = AtomicBool::new(false);

    let report = thread::scope(|s| {
        let worker = s.spawn(|| {
            while !stop.load(Ordering::SeqCst) {
                // здесь как будто автосохраняем документ
                thread::sleep(Duration::from_millis(5));
            }
            "воркер автосохранения аккуратно остановился"
        });

        thread::sleep(Duration::from_millis(30)); // main занимается другой работой
        stop.store(true, Ordering::SeqCst); // просим воркер остановиться
        worker.join().unwrap()
    });

    println!("{report}");
}

Вывод:

воркер автосохранения аккуратно остановился

▶ Попробуй сам

Воркер проверяет флаг на каждом круге цикла. Он доделывает текущий шаг и выходит сам — это гораздо бережнее, чем убить поток посреди сохранения файла.

`OnceLock` и `LazyLock`: настроить один раз

Иногда нужно одно глобальное значение, которое могут читать все потоки: таблица для поиска, имя приложения, конфигурация, загруженная при старте. Самое сложное — его создать. Что, если два потока попытаются сделать это одновременно?

Эту задачу решают std::sync::LazyLock и std::sync::OnceLock. Значение создаётся ровно один раз — в первый раз, когда оно кому-то понадобилось. Если другие потоки спросят о нём, пока оно строится, они просто подождут, а потом все будут пользоваться одним и тем же значением.

use std::collections::HashMap;
use std::sync::{LazyLock, OnceLock};
use std::thread;

// LazyLock: рецепт написан прямо здесь и выполняется при первом использовании.
static CURRENCIES: LazyLock<HashMap<&str, &str>> = LazyLock::new(|| {
    println!("строим таблицу валют...");
    HashMap::from([("US", "USD"), ("JP", "JPY"), ("DE", "EUR"), ("BR", "BRL")])
});

// OnceLock: начинает пустым, а как его заполнить, ты решаешь там, где используешь.
static STORE_NAME: OnceLock<String> = OnceLock::new();

fn store_name() -> &'static str {
    STORE_NAME.get_or_init(|| {
        std::env::var("STORE_NAME").unwrap_or_else(|_| String::from("Rusty Shop"))
    })
}

fn main() {
    let mut handles = Vec::new();
    for country in ["US", "JP", "DE", "BR"] {
        handles.push(thread::spawn(move || {
            format!("{} в стране {country} принимает оплату в {}", store_name(), CURRENCIES[country])
        }));
    }
    for handle in handles {
        println!("{}", handle.join().unwrap());
    }
}

Вывод:

строим таблицу валют...
Rusty Shop в стране US принимает оплату в USD
Rusty Shop в стране JP принимает оплату в JPY
Rusty Shop в стране DE принимает оплату в EUR
Rusty Shop в стране BR принимает оплату в BRL

▶ Попробуй сам

Четыре потока хотят получить таблицу почти одновременно, но «строим таблицу валют…» печатается только один раз. А поскольку это значения static, которые живут всю программу, потокам вообще не нужен Arc.

Что же выбрать?

LazyLock — когда ты знаешь, как построить значение, прямо в месте объявления. Он ощущается как обычная глобальная переменная.
OnceLock — когда значение определяется позже, например по флагу командной строки. Его заполняют через get_or_init(...) или set(...) (а set вежливо возвращает ошибку, если значение уже есть).

Оба входят в стандартную библиотеку. LazyLock появился в Rust 1.80, поэтому в старом коде для той же задачи часто встречаются крейты once_cell или lazy_static.

Как избежать взаимных блокировок

Компилятор не допускает гонок данных, но есть одна ошибка конкурентности, которую он поймать не может: взаимная блокировка (deadlock). Это когда потоки вечно ждут друг друга, а программа просто замирает. Ни ошибки, ни падения — ничего.

Классический рецепт — две блокировки, которые берутся в разном порядке. Представь двух людей, каждому из которых нужны и ручка, и блокнот. Один хватает ручку, другой — блокнот, и каждый вежливо ждёт, пока другой отпустит своё. Вечно.

// Поток A                           // Поток B
let from = alice.lock().unwrap();    let from = bob.lock().unwrap();
let to   = bob.lock().unwrap();      let to   = alice.lock().unwrap();
// A держит alice и ждёт bob.        // B держит bob и ждёт alice. Застряли!

Три привычки уберегут тебя:

Всегда бери блокировки в одном и том же порядке. Если каждый поток блокирует alice раньше bob (например, по возрастанию ID счёта), цикл ожидания невозможен.
Держи блокировку как можно короче. Держи её ровно столько, сколько нужно, чтобы прочитать или изменить данные. Никогда не держи её во время медленной работы — сетевого запроса, записи в файл или долгого вычисления. Всё это время все остальные потоки стоят и ждут.
Удаляй охрану пораньше. Охрана (guard) снимает блокировку, когда удаляется. Её можно завершить внутренним блоком { ... } (как у писателя RwLock выше) или вызвать drop(guard) самому.

Вот привычка 3 в действии. Каждый воркер берёт задание из общей очереди, отпускает очередь и только потом делает медленную часть:

use std::sync::Mutex;
use std::thread;

fn main() {
    let queue = Mutex::new(vec!["уменьшить фото", "отправить письмо", "собрать отчёт"]);
    let finished = Mutex::new(Vec::new());

    thread::scope(|s| {
        for _ in 0..3 {
            s.spawn(|| {
                let mut jobs = queue.lock().unwrap();
                let job = jobs.pop();
                drop(jobs); // отпускаем очередь ДО медленной части

                if let Some(job) = job {
                    let result = format!("готово: {job}"); // представь, что это занимает секунды
                    finished.lock().unwrap().push(result);
                }
            });
        }
    });

    // Область закончилась, так что `finished` больше никто не использует.
    let mut done = finished.into_inner().unwrap();
    done.sort(); // потоки заканчивают в любом порядке
    for line in done {
        println!("{line}");
    }
}

Вывод:

готово: отправить письмо
готово: собрать отчёт
готово: уменьшить фото

▶ Попробуй сам

Без drop(jobs) охрана жила бы до конца замыкания. Тогда «медленная» работа шла бы, пока очередь всё ещё заблокирована, и три воркера работали бы по очереди, а не бок о бок.

Один коварный случай, о котором стоит знать: в while let Some(job) = queue.lock().unwrap().pop() { ... } охрана живёт всё тело цикла. То есть очередь остаётся заблокированной, пока ты работаешь над заданием. Сначала возьми задание отдельной инструкцией let, как в примере.

Какой инструмент выбрать?

Ты хочешь…	Бери
Передавать работу или результаты из одного потока в другой	Канал (`mpsc`)
Дать многим потокам читать и менять одно значение	`Arc<Mutex<T>>`
Делить данные, которые часто читают и редко меняют	`Arc<RwLock<T>>`
Считать что-то или делить простой флаг «вкл/выкл»	Атомики (`AtomicUsize`, `AtomicBool`, …)
Один раз построить глобальное значение и потом только читать	`LazyLock` или `OnceLock`
Дать потокам заимствовать локальные данные без `Arc`	`thread::scope`

Всё это свободно сочетается. Примеры на этой странице уже соединяли thread::scope с атомиками и с Mutex. Выбирай самый простой подходящий инструмент и не бойся выбрать «не тот»: поменять потом обычно несложно.

Частые ошибки

Заимствовать локальную переменную в thread::spawn. Выглядит безобидно:

use std::thread;

fn main() {
    let sales = vec![12, 7, 30, 5];

    let handle = thread::spawn(|| {
        let total: i32 = sales.iter().sum();
        total
    });

    println!("итого: {}", handle.join().unwrap());
}

error[E0373]: closure may outlive the current function, but it borrows `sales`, which is owned by the current function
 --> src/main.rs:6:32
  |
6 |     let handle = thread::spawn(|| {
  |                                ^^ may outlive borrowed value `sales`
7 |         let total: i32 = sales.iter().sum();
  |                          ----- `sales` is borrowed here
  |
help: to force the closure to take ownership of `sales` (and any other referenced variables), use the `move` keyword

Компилятор говорит (по-английски), что замыкание может пережить функцию, а при этом заимствует sales, и подсказывает добавить move. Ты-то знаешь, что сразу вызываешь join(), но thread::spawn этого знать не может. С его точки зрения поток может работать вечно. Исправить можно двумя способами: добавить move (тогда main больше не сможет пользоваться sales) или использовать thread::scope, который умеет заимствовать:

let total: i32 = thread::scope(|s| s.spawn(|| sales.iter().sum()).join().unwrap());

Забыть Arc. Без Arc замыкание с move в первом потоке забирает Mutex, и следующему ничего не остаётся:

use std::sync::Mutex;
use std::thread;

fn main() {
    let counter = Mutex::new(0);
    let mut handles = vec![];

    for _ in 0..3 {
        handles.push(thread::spawn(move || {
            *counter.lock().unwrap() += 1;
        }));
    }

    for handle in handles {
        handle.join().unwrap();
    }
}

error[E0382]: use of moved value: `counter`
  --> src/main.rs:9:36
   |
 5 |     let counter = Mutex::new(0);
   |         ------- move occurs because `counter` has type `std::sync::Mutex<i32>`, which does not implement the `Copy` trait
...
 8 |     for _ in 0..3 {
   |     ------------- inside of this loop
 9 |         handles.push(thread::spawn(move || {
   |                                    ^^^^^^^ value moved into closure here, in previous iteration of loop

counter переместился в замыкание ещё на предыдущей итерации цикла. Оберни его в Arc::new(...) и клонируй Arc в начале цикла, как на странице Многопоточность. Или используй thread::scope и убери move. А если взять Rc вместо Arc, получишь error[E0277]: `Rc<std::sync::Mutex<i32>>` cannot be sent between threads safely: Rc предназначен только для одного потока (см. Умные указатели).

Практика

1. Сумма в четыре части

Создай Vec с числами от 1 до 1000. С помощью thread::scope раздели его на 4 части по 250 и посчитай сумму каждой части в своём потоке. Выведи сумму каждой части и общий итог. Подсказка: numbers.chunks(250) даёт срезы. Собери дескрипторы потоков в Vec, а потом дождись их по порядку.

Показать решение

use std::thread;

fn main() {
    let numbers: Vec<u64> = (1..=1000).collect();

    let sums: Vec<u64> = thread::scope(|s| {
        let mut handles = Vec::new();
        for chunk in numbers.chunks(250) {
            handles.push(s.spawn(move || chunk.iter().sum::<u64>()));
        }
        handles.into_iter().map(|h| h.join().unwrap()).collect()
    });

    println!("суммы частей: {:?}", sums);
    println!("итого: {}", sums.iter().sum::<u64>());
}

Вывод:

суммы частей: [31375, 93875, 156375, 218875]
итого: 500500

▶ Попробуй сам

move здесь перемещает в каждый поток только chunk — небольшой заимствованный срез. Сам вектор по-прежнему заимствуется, а не перемещается, так что после области numbers можно использовать снова.

2. Уникальные номера заказов

Интернет-магазин даёт каждому заказу номер, начиная с 1000. Запусти 4 потока, каждый из которых берёт по 3 номера заказа из одного общего AtomicU64. Собери все 12 номеров, отсортируй и выведи. Подсказка: fetch_add(1, Ordering::SeqCst) возвращает значение до прибавления, поэтому каждый номер выдаётся ровно один раз.

Показать решение

use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;

fn main() {
    let next_order = AtomicU64::new(1000);

    let mut numbers: Vec<u64> = thread::scope(|s| {
        let mut handles = Vec::new();
        for _ in 0..4 {
            handles.push(s.spawn(|| {
                (0..3)
                    .map(|_| next_order.fetch_add(1, Ordering::SeqCst))
                    .collect::<Vec<u64>>()
            }));
        }
        handles.into_iter().flat_map(|h| h.join().unwrap()).collect()
    });

    numbers.sort();
    println!("{} заказов: {:?}", numbers.len(), numbers);
}

Вывод:

12 заказов: [1000, 1001, 1002, 1003, 1004, 1005, 1006, 1007, 1008, 1009, 1010, 1011]

▶ Попробуй сам

Какой поток получит какой номер, меняется от запуска к запуску, но ни один номер не выдаётся дважды и не пропускается. Сортировка делает выводимый список одинаковым каждый раз.

Итоги

thread::scope гарантирует, что его потоки закончатся раньше него самого, поэтому они могут заимствовать локальные данные без Arc и move.
RwLock пускает много читателей или одного писателя. Он полезен для данных, которые в основном читают; в остальных случаях Mutex проще.
Атомики (AtomicUsize, AtomicI64, AtomicBool) справляются со счётчиками и флагами без блокировки. Если не уверен, бери SeqCst, для простых счётчиков — Relaxed.
LazyLock и OnceLock безопасно создают глобальное значение ровно один раз, даже если о нём одновременно спрашивают многие потоки.
Избегай взаимных блокировок: бери блокировки в одном порядке, держи их недолго и удаляй охрану пораньше.

Проверь себя

1. Почему потоки, запущенные через s.spawn внутри thread::scope, могут заимствовать локальные переменные?

2. Какое утверждение о RwLock верно?

3. Нужен счётчик просмотров страницы, общий для многих потоков. Какой простой и хороший вариант?

4. Поток A блокирует alice, потом bob. Поток B блокирует bob, потом alice. Что может случиться?

Результат: 0 / 4

Что почитать

Параллелизм с общим состоянием — книга «Язык программирования Rust» (перевод на русский): официальная глава о Mutex и Arc — основа, на которой строится эта страница.
std::thread::scope — документация стандартной библиотеки (на английском): как работают scoped-потоки, с дополнительными примерами.
std::sync::RwLock — документация стандартной библиотеки (на английском): охрана для чтения и записи, отравление (poisoning) и заметка о том, что справедливость очереди зависит от операционной системы.
std::sync::atomic — документация стандартной библиотеки (на английском): все атомарные типы и краткий обзор упорядочивания памяти.
Rust Atomics and Locks, Мара Бос (на английском): бесплатная онлайн-книга, которая шаг за шагом объясняет атомики, упорядочивание памяти и устройство блокировок.

Общее состояние в Rust: RwLock, атомики и scoped-потоки

Scoped-потоки: заимствуем вместо перемещения

RwLock: много читателей или один писатель

Атомики: счётчики без блокировки

Что такое Ordering?

Флаг остановки

OnceLock и LazyLock: настроить один раз

Как избежать взаимных блокировок

Какой инструмент выбрать?

Частые ошибки

Практика

1. Сумма в четыре части

2. Уникальные номера заказов

Итоги

Проверь себя

Что почитать

`RwLock`: много читателей или один писатель

Что такое `Ordering`?

`OnceLock` и `LazyLock`: настроить один раз