Rust mpsc Kanalları: channel, sync_channel ve İş Parçacıkları Arası Mesajlaşma

Kanal kavramı ve iki çeşit kanal

The Rust Book kanalı, içine bırakılan bir nesneyi akıntı yönünde taşıyan bir dereye benzetir: bir uçta değer bırakırsınız, öbür uçta alırsınız. Kitap aynı bölümde Go belgelerindeki "belleği paylaşarak iletişim kurmayın, iletişim kurarak belleği paylaşın" sloganını da hatırlatır. Kanalın iki ucundan biri düşürüldüğünde (drop) kanal kapanmış sayılır. Standart kütüphane iki çeşit kanal sunar:

• Asenkron, sınırsız tamponlu kanal: mpsc::channel() bir (Sender<T>, Receiver<T>) demeti döndürür. Kavramsal olarak tamponu sonsuzdur; send hiçbir zaman yer açılmasını beklemez.
• Senkron, sınırlı kanal: mpsc::sync_channel(n) bir (SyncSender<T>, Receiver<T>) döndürür. Bekleyen mesajlar önceden ayrılmış n boyutlu bir tamponda tutulur; tampon doluysa send yer açılana kadar bekler. n değeri 0 olabilir; bu durumda kanal bir buluşma (rendezvous) kanalına dönüşür ve her gönderim bir alıcıyla el sıkışmadan tamamlanmaz.

Her iki çeşitte de belge aynı sıra garantisini verir: bir göndericiden yollanan veriler alıcıda gönderildikleri sırayla görünür. Birden çok gönderici olduğunda ise farklı göndericilerin mesajlarının birbirine göre sırası, iş parçacıklarının ne zaman çalıştığına bağlıdır. std::sync modülü çok tüketicili mpmc ve tek kullanımlık oneshot kanallarını da listeler, ancak bu sürümde ikisi de deneysel (nightly-only) olarak işaretlidir; kararlı Rust'ta tek tüketicili mpsc kullanılır.

İlk kanal: send, recv ve iterator olarak Receiver

En basit kullanımda bir iş parçacığı birkaç mesaj üretir, ana iş parçacığı bunları sırayla alır. Göndericiyi kullanabilmek için iş parçacığının ona sahip olması gerekir; bu yüzden closure move ile yazılır. Alıcı bir iterator gibi gezilebilir ve tüm göndericiler düşürüldüğünde döngü kendiliğinden biter.

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

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

    thread::spawn(move || {
        for satir in ["sipariş alındı", "ödeme onaylandı", "kargoya verildi"] {
            tx.send(satir.to_string()).unwrap();
        }
        // Closure bitince tx düşürülür ve kanal kapanır.
    });

    // Receiver bir iterator gibi kullanılabilir; kanal kapanınca döngü biter.
    for mesaj in rx {
        println!("alındı: {mesaj}");
    }
    println!("kanal kapandı");
}
// çıktı: alındı: sipariş alındı
// çıktı: alındı: ödeme onaylandı
// çıktı: alındı: kargoya verildi
// çıktı: kanal kapandı

send bir Result<(), SendError<T>> döndürür. Hata yalnızca alıcı ucun artık var olmadığı durumda oluşur ve gönderdiğiniz değer hatanın içinde size geri verilir, böylece kaybolmaz. Belge burada ince bir ayrıntıya dikkat çeker: Err verinin asla alınmayacağı anlamına gelir, ama Ok verinin mutlaka alınacağı anlamına gelmez; alıcı, send başarıyla döndükten hemen sonra kapanabilir. recv ise kanalda veri yokken ve en az bir gönderici yaşarken çağıran iş parçacığını bloklar. Tüm göndericiler kapandığında Err(RecvError) döndürür; ama kanal tamponlu olduğu için kapanmadan önce gönderilmiş mesajlar yine eksiksiz teslim edilir.

Örnekteki for mesaj in rx satırı alıcının IntoIterator uygulamasını kullanır. rx.iter() de aynı davranışı ödünç alarak verir: mesaj bekler, panik yapmaz ve kanal kapandığında None ile biter. Iterator'ların genel davranışı ve collect için Rust iterator yazısına bakabilirsiniz.

send sahipliği taşır: E0382 ile yakalanan eşzamanlılık hatası

Kanal ile gönderilen değer artık alıcıya aittir. Gönderdikten sonra aynı değeri kullanmaya çalışırsanız, karşı taraf o sırada değeri değiştirmiş ya da düşürmüş olabilir. Rust bunu çalışma zamanına bırakmaz, derleme anında reddeder:

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

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

    thread::spawn(move || {
        let rapor = String::from("günlük rapor");
        tx.send(rapor).unwrap();
        println!("gönderdim: {rapor}");
    });

    println!("{}", rx.recv().unwrap());
}
// error[E0382]: borrow of moved value: `rapor`

Kitap bu hatayı, sahiplik sisteminin eşzamanlı kodda nasıl koruma sağladığının tipik örneği olarak gösterir: send parametresinin sahipliğini alır, değer taşındıktan sonra alıcı onun sahibi olur. Çözüm, değeri göndermeden önce kullanmak ya da iki tarafta da gerçekten ihtiyaç varsa bir kopyasını göndermektir:

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

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

    thread::spawn(move || {
        let rapor = String::from("günlük rapor");
        println!("gönderiyorum: {rapor}"); // kullanımı send'den önceye aldık
        tx.send(rapor).unwrap(); // sahiplik alıcıya geçer
    });

    println!("{}", rx.recv().unwrap());
}
// çıktı: gönderiyorum: günlük rapor
// çıktı: günlük rapor

Gönderilen tipin iş parçacıkları arasında taşınabilir olması da gerekir. Bunu kendimiz denedik: Rc<i32> göndermeye çalışan bir closure ile thread::spawn çağıran kod, Rc<i32> cannot be sent between threads safely mesajıyla E0277 hatası verip derlenmiyor. Send trait'inin anlamı Rust Arc ve Mutex yazısında, E0382 ailesinin genel çözümleri E0382 hata rehberi yazısında.

Birden çok üretici ve drop(tx) kuralı

Göndericiyi clone() ile çoğaltıp her üretici iş parçacığına bir kopya verirsiniz. Alıcının "artık mesaj gelmeyecek" sonucuna varması için orijinal dahil bütün göndericilerin düşürülmüş olması gerekir. En sık yapılan hata, döngüde klon dağıtıp orijinal tx değerini ana iş parçacığında tutmaya devam etmektir: işçiler biter, ama ana iş parçacığındaki gönderici yaşadığı için for döngüsü ya da recv sonsuza kadar bekler. Aşağıdaki örnekte bunu önlemek için döngüden sonra drop(tx) çağırıyoruz.

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

fn main() {
    let (tx, rx) = mpsc::channel::<(usize, u64)>();
    let dosyalar = ["a.log", "bb.log", "ccc.log", "dddd.log"];

    for (no, ad) in dosyalar.iter().enumerate() {
        let tx = tx.clone(); // her üretici kendi Sender kopyasını alır
        let ad = ad.to_string();
        thread::spawn(move || {
            let puan = ad.len() as u64 * 10; // yerine gerçek bir iş düşünün
            tx.send((no, puan)).unwrap();
        });
    }
    drop(tx); // elimizdeki son Sender: düşürmezsek döngü hiç bitmez

    let mut sonuclar: Vec<(usize, u64)> = rx.iter().collect();
    sonuclar.sort(); // geliş sırası garanti değil, sıralayıp yazdırıyoruz
    println!("{sonuclar:?}");
}
// çıktı: [(0, 50), (1, 60), (2, 70), (3, 80)]

Sonuçlar iş parçacıklarının bitiş sırasına göre gelir, bu yüzden her mesaja bir sıra numarası ekleyip toplandıktan sonra sıralıyoruz. Bu desen, işleri bölüp parçaları tek bir toplayıcıya gönderdiğiniz durumlarda (Kitap bunu birçok iş parçacığının hesapladığı parçaları tek bir iş parçacığında birleştirme örneğiyle anlatır) sonuçların deterministik olmasını sağlar. Bir iş parçacığının dönüş değerine ihtiyacınız varsa kanal yerine doğrudan JoinHandle da kullanabilirsiniz; kanal, sonuçların geldikçe işlenmesi gerektiğinde ya da üretici sayısı önceden belli olmadığında daha esnektir.

Engellemeyen ve zaman aşımlı alım: try_recv, recv_timeout

Alıcının dört temel alma yöntemi vardır. Hangi hata değerini döndürdüklerini bilmek, kanalın boş mu yoksa kapalı mı olduğunu ayırt etmenizi sağlar:

Aşağıdaki program tek iş parçacığında çalışır ama her durumu tek tek gösterir. Dikkat edilecek nokta, gönderici kapandıktan sonra tamponda kalan 8 değerinin yine de teslim edilmesi ve ancak ondan sonra hata dönmesidir. Ayrıca alıcısı düşürülmüş bir kanala yapılan send çağrısının SendError döndürdüğünü görüyorsunuz; hata tipinin Debug çıktısı içindeki değeri yazdırmaz.

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

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

    println!("{:?}", rx.try_recv()); // boş kanal, gönderen hâlâ var
    tx.send(7).unwrap();
    println!("{:?}", rx.try_recv());
    println!("{:?}", rx.recv_timeout(Duration::from_millis(20)));

    tx.send(8).unwrap();
    drop(tx); // gönderen kapandı ama tampondaki 8 kaybolmaz
    println!("{:?}", rx.recv());
    println!("{:?}", rx.recv());
    println!("{:?}", rx.try_recv());

    let (tx2, rx2) = mpsc::channel::<i32>();
    drop(rx2); // alıcı kapandı
    println!("{:?}", tx2.send(5));
}
// çıktı: Err(Empty)
// çıktı: Ok(7)
// çıktı: Err(Timeout)
// çıktı: Ok(8)
// çıktı: Err(RecvError)
// çıktı: Err(Disconnected)
// çıktı: Err(SendError { .. })

try_recv, Kitabın tarifiyle, bir iş parçacığının mesaj beklerken başka işleri de varsa kullanışlıdır: arada bir try_recv çağırıp mesaj varsa işler, yoksa başka işe devam edersiniz. Bunu sıkı bir döngüde sürekli çağırmak ise işlemciyi boşuna meşgul eder; bekleme gerçekten gerekiyorsa recv ya da recv_timeout tercih edin. Belirli bir zamana kadar beklemek için bir Instant alan recv_deadline de vardır.

sync_channel: geri basınç ve buluşma kanalı

Sınırsız kanalda üretici tüketiciden hızlıysa mesajlar bellekte birikmeye devam eder. sync_channel(n) bu durumda üreticiyi yavaşlatır: tampon dolduğunda send bekler, böylece doğal bir geri basınç (backpressure) oluşur. Beklemeden denemek için try_send kullanırsınız; tampon doluysa TrySendError::Full(değer), alıcı kapanmışsa TrySendError::Disconnected(değer) döner ve iki durumda da değer size geri verilir.

use std::sync::mpsc::{self, TrySendError};
use std::thread;

fn main() {
    // Tampon boyutu 2 olan sınırlı kanal.
    let (tx, rx) = mpsc::sync_channel::<u32>(2);

    tx.send(1).unwrap();
    tx.send(2).unwrap();
    match tx.try_send(3) {
        Err(TrySendError::Full(geri)) => println!("tampon dolu, {geri} geri verildi"),
        diger => println!("{diger:?}"),
    }
    println!("{:?}", rx.recv());

    // Kapasite 0: buluşma (rendezvous) kanalı.
    let (tx0, rx0) = mpsc::sync_channel::<&str>(0);
    let t = thread::spawn(move || {
        tx0.send("el sıkışma").unwrap(); // alıcı recv çağırana kadar bekler
        "gönderim tamam"
    });
    println!("{}", rx0.recv().unwrap());
    println!("{}", t.join().unwrap());
}
// çıktı: tampon dolu, 3 geri verildi
// çıktı: Ok(1)
// çıktı: el sıkışma
// çıktı: gönderim tamam

Kapasitesi 0 olan kanalda send, bir recv ile eşleşene kadar dönmez. Belgeye göre bu tür kanalda başarılı bir send, alıcının veriyi gerçekten aldığını garanti eder; tamponlu kanalda ise başarılı gönderim yalnızca verinin tampona yerleştiğini gösterir. Örnekte ana iş parçacığı recv çağırdığında iki taraf buluşuyor ve ancak o zaman gönderen iş parçacığı devam edip sonucunu döndürüyor.

İş havuzu deseni: tek alıcıyı işçiler arasında paylaşmak

Alıcı tek olduğu ve klonlanamadığı için "birçok işçi aynı kuyruktan iş çeksin" senaryosunda onu doğrudan paylaşamazsınız. The Rust Book'un son projesindeki iş parçacığı havuzu bu sorunu alıcıyı Arc<Mutex<Receiver<T>>> içine koyarak çözer: her işçi kilidi kısa süreliğine alır, bir iş çeker, kilidi bırakır ve işi kilitsiz yapar. Havuzu kapatmak için de göndericiyi düşürür; işçiler recv hatası alınca döngüden çıkar. Aşağıdaki örnek bu deseni küçültülmüş haliyle gösteriyor.

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

fn main() {
    let (is_tx, is_rx) = mpsc::channel::<u32>();
    let (sonuc_tx, sonuc_rx) = mpsc::channel::<(u32, u32)>();
    let is_rx = Arc::new(Mutex::new(is_rx)); // tek alıcıyı işçiler arasında paylaş

    let mut iscilar = Vec::new();
    for _ in 0..3 {
        let is_rx = Arc::clone(&is_rx);
        let sonuc_tx = sonuc_tx.clone();
        iscilar.push(thread::spawn(move || loop {
            // let ile alınan kilit, deyim bitince bırakılır; iş kilitsiz yapılır.
            let mesaj = is_rx.lock().unwrap().recv();
            match mesaj {
                Ok(n) => sonuc_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // gönderici kapandı: işçi düzgünce çıkar
            }
        }));
    }
    drop(sonuc_tx);

    for n in 1..=6 {
        is_tx.send(n).unwrap();
    }
    drop(is_tx); // iş kalmadı sinyali

    for t in iscilar {
        t.join().unwrap();
    }
    let mut sonuclar: Vec<_> = sonuc_rx.iter().collect();
    sonuclar.sort();
    println!("{sonuclar:?}");
}
// çıktı: [(1, 1), (2, 4), (3, 9), (4, 16), (5, 25), (6, 36)]

Burada kritik ayrıntı kilidin ne zaman bırakıldığıdır. Kitap, let deyiminde sağ taraftaki geçici değerlerin deyim bitince hemen düşürüldüğünü, while let, if let ve match yapılarında ise ilgili bloğun sonuna kadar yaşadığını açıklar. while let Ok(n) = is_rx.lock().unwrap().recv() biçiminde yazarsanız kilit işin tamamı boyunca tutulur ve diğer işçiler iş alamaz; kod derlenir ama paralellik kaybolur. Bu yüzden örnekte önce let mesaj = ... ile alıp sonra match yapıyoruz. Kilitlerin ayrıntısı Rust Mutex yazısında.

Ne zaman kanal, ne zaman paylaşılan durum?

Kitap kanalları tek sahipliğe benzetir: bir değeri kanaldan gönderdiğinizde artık onu kullanmazsınız. Paylaşılan durum ise çoklu sahipliğe benzer; birçok iş parçacığı aynı belleğe erişir ve bunun için kilit gerekir. İş akışı "üret, gönder, işle" biçimindeyse, örneğin bir log toplayıcı, bir iş kuyruğu ya da hesaplanan parçaların birleştirilmesi gibi, kanal genellikle daha anlaşılır kod üretir. Birçok iş parçacığının aynı yapıyı okuyup güncellemesi gerekiyorsa, örneğin ortak bir önbellek ya da sayaç, Arc, Mutex ve RwLock daha doğal bir seçimdir. İkisi birlikte de kullanılabilir: std belgesindeki Mutex örneğinde iş parçacıkları ortak sayacı kilitle artırıyor, son artışı yapan iş parçacığı da bir kanalla ana iş parçacığına haber veriyor.

Sık yapılan hatalar

• Orijinal tx'i düşürmemek: klonları dağıttıktan sonra elinizde kalan göndericiyi drop etmezseniz alıcı döngüsü hiç bitmez ve program asılı kalır.
• Gönderdikten sonra değeri kullanmak: send sahipliği alır; E0382 alırsınız. Değeri önce kullanın ya da kopyasını gönderin.
• Mesaj sırasını garanti sanmak: sıra yalnızca tek bir göndericinin kendi mesajları için korunur. Birden çok üreticide sıra numarası taşıyın.
• send sonucunu her zaman unwrap etmek: alıcı kapandıysa panik yaşarsınız. Belge birçok uygulamanın bilerek unwrap ederek hatayı iş parçacıkları arasında yaydığını söyler; uzun yaşayan servislerde ise hatayı ele alıp iş parçacığını düzgün kapatmak daha güvenlidir.
• try_recv ile meşgul bekleme: sürekli boş dönen bir döngü işlemci harcar; recv veya recv_timeout kullanın.
• Receiver'ı birden çok iş parçacığına vermeye çalışmak: alıcı tektir ve klonlanamaz. Bir iş havuzunda alıcıyı paylaşmak gerekiyorsa onu Arc<Mutex<Receiver<T>>> içine koymak bir seçenektir.

Sık Sorulan Sorular

channel ile sync_channel arasındaki fark nedir?

channel sınırsız tamponludur ve send asla yer beklemez. sync_channel(n) en fazla n bekleyen mesaj tutar, doluyken send bloklar. Üretici tüketiciden hızlı olabiliyorsa bellek kullanımını sınırlamak için sınırlı kanal tercih edin.

Kanal ne zaman kapanır?

Alıcı düşürüldüğünde ya da tüm göndericiler (klonlar dahil) düşürüldüğünde. İlk durumda send hata döndürür, ikinci durumda tampon boşaldıktan sonra recv hata döndürür ve for döngüsü biter.

Async kodda bu kanalları kullanabilir miyim?

recv çağıran iş parçacığını bloklar; bu yüzden async bir görevin içinde çağırmak yürütücüyü durdurabilir. Kitabın async bölümü, async koda uygun kanalları sağlayan çalışma zamanı (runtime) crate'lerini kullanır. Ayrıntılar Rust async/await yazısında.

Kanalla bir iş parçacığına dur komutu nasıl verilir?

En sade yol, iş parçacığının dinlediği kanalın göndericisini düşürmektir: iş parçacığı for döngüsünden çıkar ve kendiliğinden biter. Ayrı bir komut tipi gerekiyorsa mesajları bir enum ile modelleyip içine bir dur varyantı ekleyebilirsiniz; enum için Rust enum yazısına bakın.

Marka adları sahiplerine aittir. Bu yazı bağımsız bir teknik rehberdir; adı geçen kuruluşlarla bayilik veya ortaklık ilişkisi bulunduğu anlamına gelmez.

İlgili içerikler: Rust öğrenme rehberi · Rust thread::spawn · Rust Arc ve Mutex · Rust async/await · Rust iterator · Rust E0382 hatası