Каналы в Rust: передача сообщений между потоками

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

Представь канал как трубу с двумя концами. Один поток бросает значения в конец отправителя (sender). Другой поток достаёт их с конца получателя (receiver) в том же порядке. Никто ничего не делит: каждое значение переходит из одного потока в другой. В сообществе Go есть поговорка, которую Rust здесь заимствует: «Не общайтесь через общую память; вместо этого делитесь памятью, общаясь».

Твой первый канал

Канал стандартной библиотеки живёт в std::sync::mpsc. Название расшифровывается как multiple producer, single consumer — «много производителей, один потребитель»: отправителей может быть много, получатель один.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("привет от воркера");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main получил: {received}");
}

Вывод:

main получил: привет от воркера

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

Шаг за шагом:

• mpsc::channel() возвращает пару: Sender и Receiver. По традиции их называют tx (transmit, «передавать») и rx (receive, «получать»).
• move отдаёт tx новому потоку — так же, как ты перемещал данные в потоки на странице Многопоточность.
• tx.send(value) кладёт значение в канал. Он возвращает Result, потому что отправка не удастся, если получателя уже нет.
• rx.recv() ждёт, пока придёт значение. Он возвращает Err, только когда все отправители исчезли и канал пуст.

Заметь, что join() нам не понадобился. recv() и так ждёт сообщение, поэтому main не может закончиться слишком рано.

Получаем много значений

Receiver работает как итератор. Цикл for по нему сам вызывает recv() снова и снова и заканчивается сам, когда все отправители удалены:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i * 10).unwrap();
        }
        // tx удаляется здесь, когда поток заканчивается
    });

    for value in rx {
        println!("получено {value}");
    }
    println!("канал закрыт, всё готово");
}

Вывод:

получено 10
получено 20
получено 30
канал закрыт, всё готово

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

Отправитель здесь один, поэтому значения приходят ровно в том порядке, в каком их отправили. Правило «цикл останавливается, когда отправителей больше нет» очень важно. Запомни его для следующего раздела.

Несколько производителей: клонируем отправителя

Чтобы отправлять могли несколько потоков, дай каждому свою копию отправителя через tx.clone(). Все копии пишут в один и тот же канал:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("воркер {id} на связи")).unwrap();
        });
    }
    drop(tx); // удали оригинал, иначе цикл ниже никогда не закончится

    let mut messages: Vec<String> = rx.iter().collect();
    messages.sort();
    for m in messages {
        println!("{m}");
    }
}

Вывод:

воркер 1 на связи
воркер 2 на связи
воркер 3 на связи

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

На что обратить внимание:

• drop(tx) важен. Каждый поток удаляет свой клон, когда заканчивает работу, но main всё ещё держит оригинал. Пока существует хотя бы один отправитель, получатель продолжает ждать, и программа зависла бы навсегда. Удаляя оригинал, мы говорим: «main ничего отправлять не будет».
• Потоки заканчивают в любом порядке. Сообщения могут прийти как воркер 2, воркер 1, воркер 3. Мы собираем их и сортируем, чтобы вывод всегда был одинаковым. Убери sort() и запусти несколько раз — увидишь, как меняется порядок.

Не ждать вечно: try_recv и recv_timeout

recv() блокируется, пока что-нибудь не придёт. Иногда ждать не хочется. Тут помогут два родственника:

• try_recv() возвращается сразу: со значением, с Empty (пока ничего нет) или с Disconnected (все отправители исчезли).
• recv_timeout(duration) ждёт, но не дольше заданного времени. Если за это время ничего не пришло, получишь Timeout.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel::<i32>();

    // Пока ничего не отправлено: try_recv возвращается сразу.
    println!("{:?}", rx.try_recv());

    thread::spawn(move || tx.send(7).unwrap()).join().unwrap();

    // Значение уже ждёт, поэтому оно возвращается без ожидания.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // Отправителя нет, а канал пуст.
    println!("{:?}", rx.try_recv());
}

Вывод:

Err(Empty)
Ok(7)
Err(Disconnected)

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

Мы делаем join() потока до получения, поэтому значение уже ждёт в канале, и вывод всегда один и тот же.

try_recv() удобен в цикле, у которого есть и другая работа, — например, в игровом цикле, который раз в кадр проверяет новые сообщения. recv_timeout() подходит для случая «сдаёмся, если воркер работает слишком долго».

Ограниченные каналы: sync_channel

У обычного channel() нет ограничения по размеру. Если отправитель гораздо быстрее получателя, сообщения копятся в памяти. mpsc::sync_channel(n) создаёт канал, который вмещает не больше n сообщений. Когда он заполнен, send() ждёт, пока получатель не заберёт одно. Это называется обратным давлением (backpressure): медленный потребитель автоматически притормаживает быстрого производителя.

use std::sync::mpsc;
use std::thread;

fn main() {
    // Место только для 2 сообщений одновременно.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // блокируется, пока буфер полон
        }
    });

    let total: i32 = rx.iter().sum();
    producer.join().unwrap();
    println!("сумма = {total}");
}

Вывод:

сумма = 15

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

Производитель никогда не может обогнать main больше чем на два числа. С sync_channel(0) буфера нет вообще: каждый send() ждёт, пока получатель будет готов принять значение, — как будто передаёшь посылку из рук в руки.

Маленький пул воркеров

Теперь соберём всё вместе. Пул воркеров (worker pool) — это несколько потоков, которые берут задания из общей очереди и отправляют обратно результаты. Здесь используются два канала: один для заданий, другой для результатов.

Есть одна загвоздка: Receiver нельзя клонировать (потребитель ведь один). Поэтому воркеры делят получатель заданий через Arc<Mutex<_>> и по очереди достают из него задания:

use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();

    // Получатели не клонируются, поэтому воркеры делят один через Arc<Mutex<_>>.
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _ in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(thread::spawn(move || loop {
            let job = job_rx.lock().unwrap().recv();
            match job {
                Ok(n) => result_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // заданий больше нет
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // говорит воркерам, что работы больше не будет

    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    for w in workers {
        w.join().unwrap();
    }
    results.sort();
    for (n, square) in results {
        println!("{n} в квадрате = {square}");
    }
}

Вывод:

1 в квадрате = 1
2 в квадрате = 4
3 в квадрате = 9
4 в квадрате = 16
5 в квадрате = 25
6 в квадрате = 36

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

Как пул аккуратно завершается:

1. drop(job_tx) закрывает канал заданий. После этого recv() у каждого воркера возвращает Err, и воркер выходит из цикла.
2. Когда воркер заканчивает, удаляется его клон result_tx. Оригинал мы уже удалили, поэтому, как только все три воркера закончат, канал результатов закроется.
3. Тогда result_rx.iter() останавливается, а результаты мы сортируем, потому что воркеры заканчивают в любом порядке.

Небольшая деталь: job_rx.lock().unwrap().recv() написан в одной инструкции let намеренно. Блокировка снимается в конце этой инструкции, поэтому другие воркеры могут взять следующее задание, пока этот ещё возводит своё число в квадрат.

Каналы или Arc<Mutex<T>>

Оба способа безопасны. Просто они подходят для задач разной формы:

Используй канал, когда…	Используй Arc<Mutex<T>>, когда…
Работа течёт в одну сторону: производители создают, потребитель обрабатывает	Многим потокам нужно читать и обновлять одно и то же значение (счётчик, кеш)
Нужна очередь заданий, событий или строк лога	Данные большие, и копировать или перемещать их было бы расточительно
У каждого значения в каждый момент один владелец	Потокам нужно смотреть на текущее состояние, а не на поток обновлений

Хорошее практическое правило: если программу можно описать как «эти потоки сообщают тому потоку что-то», используй канал. Если «все эти потоки редактируют одну вещь» — Mutex. Многие реальные программы используют и то и другое, как наш пул воркеров.

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

Использовать значение после отправки. send() забирает владение — так же, как передача значения в функцию. После отправки значение принадлежит получателю:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("hi");
        tx.send(message).unwrap();
        println!("sent {message}");
    });

    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `message`
  --> src/main.rs:10:25
   |
 8 |         let message = String::from("hi");
   |             ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(message).unwrap();
   |                 ------- value moved here
10 |         println!("sent {message}");
   |                         ^^^^^^^ value borrowed here after move

Компилятор сообщает (по-английски), что message заимствуется в строке 10 уже после того, как была перемещена в строке 9. Так Rust тебя защищает: если бы отправитель мог и дальше менять message, оба потока трогали бы одни и те же данные. Выводи значение до отправки или отправляй .clone().

Перемещать отправителя в цикл с потоками. Первый поток забирает tx, и второму ничего не остаётся:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        thread::spawn(move || {
            tx.send(id).unwrap();
        });
    }

    for value in rx {
        println!("{value}");
    }
}

error[E0382]: use of moved value: `tx`
  --> src/main.rs:8:23
   |
 5 |     let (tx, rx) = mpsc::channel();
   |          -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
...
 8 |         thread::spawn(move || {
   |                       ^^^^^^^ value moved into closure here, in previous iteration of loop

Здесь компилятор говорит, что tx был перемещён в замыкание ещё на предыдущей итерации цикла. Исправление — let tx = tx.clone(); в начале цикла, как в примере с несколькими производителями.

Программа, которая никогда не заканчивается. Тут нет ошибки компиляции, и поэтому это коварно. Если перебирать получатель в цикле (for x in rx), пока жив хоть один отправитель — часто это оригинальный tx в main, — цикл будет ждать вечно. Когда программа «просто зависла», проверь, что каждый отправитель удаляется.

Что дальше

std::sync::mpsc покрывает многое, но тебе встретятся две популярные альтернативы:

• crossbeam-channel (на английском) добавляет каналы с несколькими потребителями (получатели, которые можно клонировать) и макрос select!, чтобы ждать сразу несколько каналов.
• tokio::sync::mpsc — канал для асинхронного кода. Если ты используешь async/.await (см. Async/Await), бери его вместо стандартного: его recv().await не блокирует весь поток.

Итоги

• mpsc::channel() даёт пару (Sender, Receiver). send() перемещает значение внутрь, recv() ждёт следующее.
• Клонируй отправителя, чтобы производителей было много. Получатель всегда один.
• Цикл по получателю заканчивается, когда удалены все отправители, поэтому не забудь сделать drop() для оригинала.
• try_recv() и recv_timeout() позволяют проверить канал, не ожидая вечно. sync_channel(n) ограничивает очередь n сообщениями.
• Сообщения от нескольких потоков приходят в любом порядке: сортируй или суммируй их, если нужен стабильный вывод.
• Используй каналы, когда данные текут от потока к потоку. Используй Arc<Mutex<T>>, когда потоки делят и редактируют одно значение.

Проверь себя

Проверь себя

1. Как расшифровывается «mpsc»?

Many processes, shared cache Message passing, safe concurrency Multiple producer, single consumer Mutex protected shared channel

2. Как дать трём потокам отправлять в один и тот же канал?

Дать каждому потоку tx.clone() Обернуть получатель в Rc Вызвать mpsc::channel() три раза и объединить каналы Никак: у канала только один отправитель

3. Цикл for value in rx никогда не заканчивается. Какая причина самая вероятная?

Канал переполнен Какой-то отправитель (часто оригинальный tx в main) так и не был удалён recv() был вызван слишком рано Значения не реализуют Copy

4. Что делает send(), когда в sync_channel(2) уже лежат 2 сообщения?

Удаляет самое старое сообщение Сразу возвращает ошибку Паникует Ждёт, пока получатель не заберёт сообщение

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

Что почитать

• Пересылка сообщений для передачи данных между потоками — книга «Язык программирования Rust» (перевод на русский): официальное введение в каналы с дополнительными примерами.
• std::sync::mpsc — документация стандартной библиотеки (на английском): все методы Sender, SyncSender и Receiver.
• Каналы — Rust на примерах (перевод на русский): короткий запускаемый пример с несколькими потоками-производителями.
• Превращение нашего однопоточного сервера в многопоточный сервер — книга «Язык программирования Rust» (перевод на русский): шаг за шагом строит полноценный пул потоков на каналах и Arc<Mutex<Receiver>>.
• tokio::sync::mpsc — docs.rs (на английском): асинхронный канал для Tokio и чем он отличается от стандартного.