В Rust mpsc (multi-producer, single-consumer) — это один из ключевых механизмов межпоточной коммуникации, позволяющий потокам обмениваться данными безопасно и без общего состояния, то есть без необходимости использовать мьютексы. Каналы реализованы в стандартной библиотеке в модуле std::sync::mpsc, и основаны на концепции message passing: данные не разделяются между потоками, а передаются как сообщения — с владением, строго и безопасно.

Основная идея

mpsc расшифровывается как multi-producer, single-consumer:

• Multi-producer — можно создать несколько передающих сторон (потоков), каждый из которых может отправлять данные.

• Single-consumer — только один поток может принимать данные.

Это соответствует паттерну, при котором много потоков могут «писать» в канал, а один поток «читает» оттуда.

Объявление и структура

Канал состоит из двух частей:

• Sender<T> — отправитель (tx, «transmitter»)

• Receiver<T> — получатель (rx)

Создаются они с помощью функции channel():

use std::sync::mpsc;
let (tx, rx) = mpsc::channel();

Здесь:

• tx — объект, через который отправляются значения типа T;

• rx — объект, через который эти значения можно получить.

Простой пример использования

use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
tx.send("Привет из потока!").unwrap();
});
let msg = rx.recv().unwrap();
println!("Получено: {}", msg);
}

• send передаёт данные.

• recv блокирует текущий поток до получения данных.

• Метод unwrap() здесь используется для простоты — на практике желательно обрабатывать ошибки.

Несколько отправителей (multi-producer)

Чтобы несколько потоков могли отправлять данные через один канал, нужно клонировать Sender:

use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
for i in 0..3 {
let tx_clone = tx.clone();
thread::spawn(move || {
tx_clone.send(format!("Сообщение от потока {}", i)).unwrap();
});
}
for _ in 0..3 {
let msg = rx.recv().unwrap();
println!("Получено: {}", msg);
}
}

Здесь:

• tx.clone() создаёт независимую копию отправителя;

• все Sender-ы работают с общим Receiver;

• Receiver по-прежнему один — читать сообщения можно только из одного потока.

Поведение при закрытии канала

Канал закрывается, когда все Sender уничтожены (дропнуты). Тогда recv() или try_recv() на Receiver вернёт Err(RecvError).

Пример:

let (tx, rx) = mpsc::channel();
drop(tx); // канал закрыт
match rx.recv() {
Ok(val) => println!("{}", val),
Err(\_) => println!("Канал закрыт, данных нет."),
}

Методы получения данных

1. recv() — блокирующий, ждёт, пока придут данные или канал не закроется.

2. try_recv() — неблокирующий, сразу возвращает Result, если ничего нет — ошибка.

3. iter() — удобный способ получать данные в цикле до тех пор, пока канал не закрыт.

for val in rx {
println!("Получено: {}", val);
}

Передача владения

Когда вы отправляете значение через канал, владение передаётся:

let (tx, rx) = mpsc::channel();
let msg = String::from("Привет");
tx.send(msg).unwrap();
// println!("{}", msg); // Ошибка: владение msg передано!

Rust на уровне компиляции гарантирует, что данные не будут использоваться после передачи.

Использование с Arc<Mutex

Иногда, если нужно нескольким потокам и отправлять, и принимать, можно использовать Arc<Mutex + channel вместе, но чаще предпочтительнее сохранять одностороннюю передачу владения через mpsc.

Когда использовать mpsc в Rust

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

• Когда вы хотите избежать гонки данных без использования Mutex.

• Когда проще передавать владение данными, чем синхронизировать доступ к ним.

Ограничения std::sync::mpsc

1. Только один потребитель — если вам нужно несколько, используйте библиотеку crossbeam (crossbeam_channel).

2. Не подходит для асинхронного кода — в таком случае используйте tokio::sync::mpsc или async_channel.

3. Не поддерживает буферизацию (без sync_channel), т.е. channel() — это небуферизованный канал (но можно сделать буферизованный через sync_channel).

Буферизованные каналы

Вы можете использовать sync_channel(capacity) для создания ограниченного буферизированного канала:

let (tx, rx) = mpsc::sync_channel(2);

Здесь send() будет блокировать поток, если буфер переполнен, пока потребитель не освободит место.

mpsc каналы — один из важнейших инструментов Rust для безопасного и производительного межпоточного взаимодействия, при этом они интуитивно понятны и строго следуют модели владения языка.