add chapter 08

Author: smallnest

.DS_Store

@@ -4,6 +4,7 @@ pub use others::*;
 use std::sync::mpsc;
 use std::sync::mpsc::sync_channel;
 use std::thread;
+use std::time::Duration;
 
 pub fn mpsc_example1() {
     let (tx, rx) = mpsc::channel();
@@ -60,3 +61,27 @@ pub fn mpsc_example4() {
 
     println!("mpsc_example4 completed");
 }
+
+pub fn mpsc_drop_example() {
+        // 创建一个有边界的多生产者、单消费者的通道
+        let (sender, receiver) = mpsc::channel::<i32>(); // 指定通道中传递的数据类型为 i32
+
+        // 启动三个生产者线程
+        for i in 0..3 {
+            let tx = sender.clone(); // 克隆发送端，每个线程都拥有独立的发送端
+            thread::spawn(move || {
+                thread::sleep(Duration::from_secs(1)); // 等待所有线程启动完毕
+                tx.send(i).expect("Failed to send message");
+            });
+        }
+        
+        
+        // 丢弃发送端，不影响clone
+        drop(sender); 
+      
+    
+        // 主线程作为消费者，接收来自生产者线程的消息
+        for received_message in receiver {
+            println!("Received message: {}", received_message);
+        }
+}

book_cn/rust_concurrency_cookbook.pdf

@@ -5,14 +5,20 @@ fn main() {
     mpsc_example2();
     mpsc_example3();
     mpsc_example4();
+    mpsc_drop_example();
 
     crossfire_mpsc();
     crossfire_mpmc();
+
     flume_example();
+    flume_select();
+    flume_async();
+
     async_channel_example();
     async_priority_channel_example();
     futures_channel_mpsc_example();
     futures_channel_oneshot_example();
     kanal_example();
     kanal_async_example();
+    kanal_oneshot_example();
 }

channel/src/lib.rs

@@ -74,8 +74,6 @@ pub fn crossfire_mpmc() {
     });
 }
 
-
-
 // has issues.
 pub fn broadcaster() {
     // let rt = tokio::runtime::Runtime::new().unwrap();
@@ -107,6 +105,35 @@ pub fn flume_example() {
     assert_eq!((0..10).sum::<u32>(), received);
 }
 
+pub fn flume_select() {
+    let (tx0, rx0) = flume::unbounded();
+    let (tx1, rx1) = flume::unbounded();
+
+    std::thread::spawn(move || {
+        tx0.send(true).unwrap();
+        tx1.send(42).unwrap();
+    });
+
+    flume::Selector::new()
+        .recv(&rx0, |b| println!("Received {:?}", b))
+        .recv(&rx1, |n| println!("Received {:?}", n))
+        .wait();
+}
+
+pub fn flume_async() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    let (tx, rx) = flume::unbounded();
+
+    rt.block_on(async move {
+        tokio::spawn(async move {
+            tx.send_async(5).await.unwrap();
+        });
+
+        println!("flume async rx: {}", rx.recv_async().await.unwrap());
+    });
+}
+
 pub fn async_channel_example() {
     let rt = tokio::runtime::Runtime::new().unwrap();
 
@@ -194,7 +221,7 @@ pub fn kanal_example() {
     });
 
     let received: u32 = rx.sum();
-    
+
     println!("received sum: {}", received);
 }
 
@@ -210,4 +237,14 @@ pub fn kanal_async_example() {
 
         println!("rx: {}", rx.recv().await.unwrap());
     });
+}
+
+pub fn kanal_oneshot_example() {
+    let (tx, rx) = kanal::oneshot();
+
+    thread::spawn(move || {
+        tx.send(5).unwrap();
+    });
+
+    println!("kanal oneshot rx: {}", rx.recv().unwrap());
 }