Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

การประยุกต์ใช้คอนเคอร์เรนซีด้วย Async

ในหัวข้อนี้ เราจะประยุกต์ใช้ async กับความท้าทายด้านคอนเคอร์เรนซีแบบเดียวกับที่เราจัดการด้วยเธรดในบทที่ 16 เนื่องจากเราได้พูดถึงแนวคิดหลักจำนวนมากไปแล้วที่นั่น ในหัวข้อนี้เราจะเน้นไปที่สิ่งที่แตกต่างกันระหว่างเธรดและฟิวเจอร์ส

ในหลายๆ กรณี API สำหรับการทำงานกับคอนเคอร์เรนซีโดยใช้ async จะคล้ายคลึงอย่างมากกับ API สำหรับการใช้เธรด ในกรณีอื่นๆ พวกมันกลับแตกต่างกันค่อนข้างมาก แม้ว่า API จะ ดู คล้ายกันระหว่างเธรดและ async แต่มันมักจะมีพฤติกรรมที่แตกต่างกัน—และเกือบจะเสมอไปที่มีลักษณะด้านประสิทธิภาพที่แตกต่างกัน

การสร้างแทสก์ใหม่ด้วย spawn_task

การทำงานแรกที่เราจัดการในหัวข้อ “การสร้างเธรดใหม่ด้วย spawn” ในบทที่ 16 คือการนับตัวเลขบนสองเธรดที่แยกกัน ลองมาทำสิ่งเดียวกันโดยใช้ async กัน เครต trpl จัดหาฟังก์ชัน spawn_task ซึ่งมีลักษณะคล้ายกับ API thread::spawn อย่างมาก และจัดหาฟังก์ชัน sleep ซึ่งเป็นเวอร์ชัน async ของ API thread::sleep เราสามารถใช้สิ่งเหล่านี้ร่วมกันเพื่อสร้างตัวอย่างการนับเลข ดังแสดงในโค้ดตัวอย่างที่ 17-6

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        trpl::spawn_task(async {
            for i in 1..10 {
                println!("hi number {i} from the first task!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        });

        for i in 1..5 {
            println!("hi number {i} from the second task!");
            trpl::sleep(Duration::from_millis(500)).await;
        }
    });
}

ในฐานะจุดเริ่มต้น เราตั้งค่าฟังก์ชัน main ด้วย trpl::block_on เพื่อให้ฟังก์ชันระดับบนสุดของเราสามารถเป็น async ได้

หมายเหตุ: ตั้งแต่จุดนี้เป็นต้นไปในบทนี้ ทุกๆ ตัวอย่างจะรวมโค้ดห่อหุ้ม trpl::block_on ใน main แบบเดียวกันนี้เป๊ะๆ ดังนั้นเราจะมักจะข้ามมันไปเช่นเดียวกับที่เราข้าม main อย่าลืมใส่ลงในโค้ดของคุณด้วย!

จากนั้นเราเขียนสองลูปภายในบล็อกนั้น โดยแต่ละลูปบรรจุการเรียกใช้ trpl::sleep ซึ่งรอเป็นเวลาครึ่งวินาที (500 มิลลิวินาที) ก่อนที่จะส่งข้อความถัดไป เราใส่ลูปหนึ่งไว้ในเนื้อหาของ trpl::spawn_task และอีกลูปหนึ่งไว้ในลูป for ระดับบนสุด เรายังเพิ่ม await ไว้หลังการเรียกใช้ sleep ด้วย

โค้ดนี้มีพฤติกรรมคล้ายกับเวอร์ชันที่ใช้เธรด—รวมถึงความจริงที่ว่าคุณอาจเห็นข้อความปรากฏในลำดับที่แตกต่างกันในเทอร์มินัลของคุณเมื่อรันมัน:

hi number 1 from the second task!
hi number 1 from the first task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!

เวอร์ชันนี้จะหยุดลงทันทีที่ลูป for ในเนื้อหาของบล็อก async หลักทำงานเสร็จสิ้น เนื่องจากแทสก์ที่สร้างขึ้นโดย spawn_task จะถูกปิดเมื่อฟังก์ชัน main จบลง หากคุณต้องการให้มันรันจนกระทั่งแทสก์เสร็จสมบูรณ์ คุณจำเป็นต้องใช้จอยแฮนเดิล (join handle) เพื่อรอให้แทสก์แรกทำงานเสร็จ สำหรับเธรด เราใช้เมธอด join เพื่อ “บล็อก” จนกว่าเธรดจะทำงานเสร็จ ในโค้ดตัวอย่างที่ 17-7 เราสามารถใช้ await เพื่อทำสิ่งเดียวกันได้ เนื่องจากแทสก์แฮนเดิลเองก็เป็นฟิวเจอร์ โดยชนิด Output ของมันคือ Result ดังนั้นเราจึง unwrap มันหลังจาก await แล้วด้วย

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let handle = trpl::spawn_task(async {
            for i in 1..10 {
                println!("hi number {i} from the first task!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        });

        for i in 1..5 {
            println!("hi number {i} from the second task!");
            trpl::sleep(Duration::from_millis(500)).await;
        }

        handle.await.unwrap();
    });
}

เวอร์ชันอัปเดตนี้จะรันจนกระทั่ง ทั้งสอง ลูปทำงานเสร็จสิ้น:

hi number 1 from the second task!
hi number 1 from the first task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!
hi number 6 from the first task!
hi number 7 from the first task!
hi number 8 from the first task!
hi number 9 from the first task!

จนถึงตอนนี้ ดูเหมือนว่า async และเธรดจะให้ผลลัพธ์ที่คล้ายกัน เพียงแต่ใช้ไวยากรณ์ต่างกัน: นั่นคือใช้ await แทนที่จะเรียกใช้ join บนจอยแฮนเดิล และรอคอยการเรียกใช้ sleep

ข้อแตกต่างที่ใหญ่กว่าคือเราไม่จำเป็นต้องสร้างเธรดของระบบปฏิบัติการขึ้นมาอีกเธรดเพื่อทำสิ่งนี้ ในความเป็นจริง เราไม่จำเป็นต้องสร้างแทสก์ด้วยซ้ำในที่นี้ เนื่องจากบล็อก async ถูกแปลเป็นฟิวเจอร์สนิรนาม เราจึงสามารถใส่แต่ละลูปไว้ในบล็อก async และให้รันไทม์รันพวกมันทั้งคู่จนเสร็จสมบูรณ์โดยใช้ฟังก์ชัน trpl::join

ในหัวข้อ “การรอให้เธรดทั้งหมดทำงานเสร็จสิ้น” ในบทที่ 16 เราแสดงวิธีใช้เมธอด join บนชนิดข้อมูล JoinHandle ที่คืนค่าเมื่อคุณเรียกใช้ std::thread::spawn ฟังก์ชัน trpl::join ก็คล้ายคลึงกัน แต่ใช้สำหรับฟิวเจอร์ส เมื่อคุณมอบฟิวเจอร์สสองตัวให้มัน มันจะสร้างฟิวเจอร์ใหม่ตัวเดียวขึ้นมาซึ่งมีผลลัพธ์เป็นทูเพิลบรรจุผลลัพธ์ของฟิวเจอร์แต่ละตัวที่คุณส่งเข้าไปเมื่อพวกมันทำงานเสร็จ ทั้งคู่ ดังนั้น ในโค้ดตัวอย่างที่ 17-8 เราใช้ trpl::join เพื่อรอให้ทั้ง fut1 และ fut2 ทำงานเสร็จ เรา ไม่ได้ รอคอย fut1 และ fut2 แต่รอคอยฟิวเจอร์ใหม่ที่สร้างขึ้นโดย trpl::join แทน เราละเลยผลลัพธ์ เพราะมันเป็นเพียงทูเพิลที่บรรจุค่ายูนิตสองค่า

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let fut1 = async {
            for i in 1..10 {
                println!("hi number {i} from the first task!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let fut2 = async {
            for i in 1..5 {
                println!("hi number {i} from the second task!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        trpl::join(fut1, fut2).await;
    });
}

เมื่อเรารันสิ่งนี้ เราจะเห็นฟิวเจอร์สทั้งสองรันจนเสร็จสมบูรณ์:

hi number 1 from the first task!
hi number 1 from the second task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!
hi number 6 from the first task!
hi number 7 from the first task!
hi number 8 from the first task!
hi number 9 from the first task!

คราวนี้ คุณจะเห็นลำดับเดียวกันเป๊ะๆ ในทุกๆ ครั้ง ซึ่งแตกต่างอย่างมากจากสิ่งที่เราเห็นกับเธรดและกับ trpl::spawn_task ในโค้ดตัวอย่างที่ 17-7 นั่นเป็นเพราะฟังก์ชัน trpl::join มีความ ยุติธรรม (fair) หมายความว่ามันตรวจสอบฟิวเจอร์แต่ละตัวบ่อยเท่ากัน สลับกันไปมา และไม่ยอมให้ตัวหนึ่งแซงหน้าไปหากอีกตัวหนึ่งพร้อม สำหรับเธรด ระบบปฏิบัติการจะตัดสินใจว่าจะตรวจสอบเธรดใดและปล่อยให้รันนานแค่ไหน ส่วน async ใน Rust รันไทม์จะเป็นผู้ตัดสินใจว่าจะตรวจสอบแทสก์ใด (ในทางปฏิบัติ รายละเอียดจะซับซ้อนขึ้นเนื่องจากรันไทม์ async อาจใช้เธรดระบบปฏิบัติการอยู่เบื้องหลังเป็นส่วนหนึ่งของการจัดการคอนเคอร์เรนซี ดังนั้นการรับประกันความยุติธรรมจึงอาจเป็นงานที่หนักขึ้นสำหรับรันไทม์—แต่มันยังคงเป็นไปได้!) รันไทม์ไม่จำเป็นต้องรับประกันความยุติธรรมสำหรับการทำงานใดๆ และพวกมันมักจัดหา API ที่ต่างกันเพื่อให้คุณเลือกว่าต้องการความยุติธรรมหรือไม่

ลองทดสอบรูปแบบต่างๆ ของการรอคอยฟิวเจอร์สและดูว่าพวกมันทำอะไร:

  • นำบล็อก async ออกจากรอบๆ ลูปใดลูปหนึ่งหรือทั้งสองลูป
  • รอคอย (await) แต่ละบล็อก async ทันทีหลังจากนิยามมัน
  • ห่อหุ้มเฉพาะลูปแรกไว้ในบล็อก async และรอคอยฟิวเจอร์ที่เป็นผลลัพธ์หลังจากเนื้อหาของลูปที่สอง

สำหรับความท้าทายพิเศษ ลองดูว่าคุณสามารถหาสิ่งที่จะเป็นผลลัพธ์ในแต่ละกรณีได้ ก่อน ที่จะรันโค้ดหรือไม่!

การส่งข้อมูลระหว่างสองแทสก์โดยใช้การส่งข้อความ

การแบ่งปันข้อมูลระหว่างฟิวเจอร์สจะเป็นสิ่งที่คุณคุ้นเคยเช่นกัน: เราจะใช้การส่งข้อความ (message passing) อีกครั้ง แต่มันจะเป็นเวอร์ชัน async ของชนิดข้อมูลและฟังก์ชัน เราจะใช้แนวทางที่แตกต่างจากที่เราทำในหัวข้อ “การถ่ายโอนข้อมูลระหว่างเธรดด้วยการส่งข้อความ” ในบทที่ 16 เล็กน้อย เพื่อแสดงข้อแตกต่างหลักบางประการระหว่างคอนเคอร์เรนซีแบบอิงเธรดและแบบอิงฟิวเจอร์ส ในโค้ดตัวอย่างที่ 17-9 เราจะเริ่มด้วยบล็อก async เพียงบล็อกเดียว—โดย ไม่ สร้างแทสก์แยกต่างหากเหมือนกับที่เราเคยสร้างเธรดแยกต่างหาก

extern crate trpl; // required for mdbook test

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let val = String::from("hi");
        tx.send(val).unwrap();

        let received = rx.recv().await.unwrap();
        println!("received '{received}'");
    });
}

ในที่นี้ เราใช้ trpl::channel ซึ่งเป็นเวอร์ชัน async ของ API แชนเนลแบบผู้ผลิตหลายราย ผู้บริโภครายเดียวที่เราใช้กับเธรดในบทที่ 16 เวอร์ชัน async ของ API แตกต่างจากเวอร์ชันอิงเธรดเพียงเล็กน้อย: มันใช้ผู้รับ rx แบบเปลี่ยนแปลงได้ (mutable) แทนที่จะเป็นแบบเปลี่ยนแปลงไม่ได้ และเมธอด recv ของมันสร้างฟิวเจอร์ที่เราจำเป็นต้องรอคอย (await) แทนที่จะสร้างค่าโดยตรง คราวนี้เราสามารถส่งข้อความจากผู้ส่งไปยังผู้รับได้ สังเกตว่าเราไม่จำเป็นต้องสร้างเธรดแยกต่างหาก หรือแม้กระทั่งแทสก์แยกต่างหาก เราเพียงต้องรอคอยการเรียกใช้ rx.recv

เมธอดแบบซิงโครนัส Receiver::recv ใน std::mpsc::channel จะบล็อกจนกว่าจะได้รับข้อความ แต่เมธอด trpl::Receiver::recv ไม่ทำเช่นนั้น เพราะมันเป็น async แทนที่จะบล็อก มันจะส่งการควบคุมกลับไปยังรันไทม์จนกว่าจะได้รับข้อความ หรือฝั่งส่งของแชนเนลถูกปิดลง ในทางตรงกันข้าม เราไม่ได้รอคอยการเรียกใช้ send เพราะมันไม่ได้บล็อก มันไม่จำเป็นต้องบล็อก เพราะแชนเนลที่เราส่งลงไปเป็นแบบไม่จำกัดขอบเขต (unbounded)

หมายเหตุ: เนื่องจากโค้ด async ทั้งหมดนี้รันอยู่ในบล็อก async ในการเรียกใช้ trpl::block_on ทุกอย่างภายในนั้นจึงสามารถหลีกเลี่ยงการบล็อกได้ อย่างไรก็ตาม โค้ดที่อยู่ ภายนอก มันจะบล็อกตรงการคืนค่าฟังก์ชัน block_on นั่นคือวัตถุประสงค์ทั้งหมดของฟังก์ชัน trpl::block_on: มันช่วยให้คุณ เลือก ได้ว่าจะบล็อกตรงจุดใดในชุดโค้ด async และช่วยสร้างจุดเชื่อมต่อระหว่างโค้ด sync และ async

สังเกตสองสิ่งเกี่ยวกับตัวอย่างนี้ ประการแรก ข้อความจะมาถึงทันที ประการที่สอง แม้ว่าเราจะใช้ฟิวเจอร์ที่นี่ แต่ก็ยังไม่มีคอนเคอร์เรนซี ทุกอย่างในตัวอย่างนี้เกิดขึ้นตามลำดับ เหมือนกับที่มันจะเป็นหากไม่มีฟิวเจอร์สเข้ามาเกี่ยวข้อง

ลองมาจัดการส่วนแรกโดยการส่งชุดข้อความและหยุดพักระหว่างข้อความ ดังแสดงในโค้ดตัวอย่างที่ 17-10

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("future"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            trpl::sleep(Duration::from_millis(500)).await;
        }

        while let Some(value) = rx.recv().await {
            println!("received '{value}'");
        }
    });
}

นอกเหนือจากการส่งข้อความแล้ว เราจำเป็นต้องรับข้อความด้วย ในกรณีนี้ เนื่องจากเรารู้ว่ามีกี่ข้อความส่งเข้ามา เราสามารถทำได้ด้วยตนเองโดยเรียกใช้ rx.recv().await สี่ครั้ง ทว่าในโลกความเป็นจริง เรามักจะรอข้อความที่มีจำนวน ไม่ทราบแน่ชัด ดังนั้นเราจึงจำเป็นต้องรอต่อไปเรื่อยๆ จนกว่าจะพิจารณาได้ว่าไม่มีข้อความเหลืออยู่อีกแล้ว

ในโค้ดตัวอย่างที่ 16-10 เราใช้ลูป for เพื่อประมวลผลไอเทมทั้งหมดที่ได้รับจากแชนเนลแบบซิงโครนัส อย่างไรก็ตาม Rust ยังไม่มีวิธีใช้ลูป for กับชุดไอเทมที่ สร้างขึ้นแบบอะซิงโครนัส ดังนั้นเราจึงจำเป็นต้องใช้ลูปที่เรายังไม่เคยเห็นมาก่อน: นั่นคือลูปแบบมีเงื่อนไข while let สิ่งนี้คือเวอร์ชันลูปของโครงสร้าง if let ที่เราเคยเห็นในหัวข้อ “การควบคุมการทำงานอย่างกระชับด้วย if let และ let...else” ในบทที่ 6 ลูปจะทำงานต่อไปเรื่อยๆ ตราบใดที่รูปแบบที่มันระบุยังคงตรงกับค่า

การเรียกใช้ rx.recv จะสร้างฟิวเจอร์ขึ้นมา ซึ่งเรารอคอย รันไทม์จะหยุดพักฟิวเจอร์จนกว่ามันจะพร้อม เมื่อข้อความมาถึง ฟิวเจอร์จะคลี่คลายค่าออกมาเป็น Some(message) บ่อยเท่าที่มีข้อความมาถึง เมื่อแชนเนลปิดลง ไม่ว่าจะมีข้อความส่งมาถึง เลยหรือไม่ ฟิวเจอร์จะคลี่คลายค่าออกมาเป็น None เพื่อระบุว่าไม่มีค่าใดๆ อีกแล้ว ดังนั้นเราจึงควรหยุดโพลลิง—กล่าวคือ หยุดรอคอย

ลูป while let จะรวบรวมทั้งหมดนี้เข้าด้วยกัน หากผลลัพธ์ของการเรียกใช้ rx.recv().await คือ Some(message) เราจะสามารถเข้าถึงข้อความนั้นและใช้งานมันในเนื้อหาของลูปได้ เช่นเดียวกับที่ทำได้ด้วย if let หากผลลัพธ์เป็น None ลูปจะจบลง ทุกครั้งที่ลูปทำงานครบรอบ มันจะชนกับจุดรออีกครั้ง ดังนั้นรันไทม์จึงหยุดพักมันอีกครั้งจนกว่าข้อความอื่นจะมาถึง

ตอนนี้โค้ดส่งและรับข้อความทั้งหมดได้สำเร็จแล้ว แต่น่าเสียดายที่ยังคงมีปัญหาอยู่อีกสองสามประการ ประการหนึ่งคือ ข้อความไม่ได้มาถึงในช่วงเวลาเว้นห่างกันครึ่งวินาที พวกมันมาถึงพร้อมกันทั้งหมด 2 วินาที (2,000 มิลลิวินาที) หลังจากที่เราเริ่มโปรแกรม อีกประการหนึ่งคือ โปรแกรมนี้ไม่มีวันจบการทำงาน! แต่กลับรอข้อความใหม่ตลอดไป คุณจะต้องปิดมันโดยใช้ ctrl-C

โค้ดภายในบล็อก Async บล็อกเดียวจะทำงานเป็นเส้นตรง

ลองเริ่มจากการตรวจสอบว่าทำไมข้อความจึงมาถึงพร้อมกันทั้งหมดหลังจากการหยุดพักเต็มเวลา แทนที่จะมาถึงพร้อมการหยุดพักระหว่างแต่ละข้อความ ภายในบล็อก async บล็อกหนึ่ง ลำดับที่คีย์เวิร์ด await ปรากฏในโค้ดก็คือลำดับที่พวกมันถูกประมวลผลเมื่อโปรแกรมรัน

มีบล็อก async เพียงบล็อกเดียวในโค้ดตัวอย่างที่ 17-10 ดังนั้นทุกอย่างในนั้นจึงรันเป็นเส้นตรง และยังไม่มีคอนเคอร์เรนซี การเรียกใช้ tx.send ทั้งหมดเกิดขึ้น สลับกับการเรียกใช้ trpl::sleep ทั้งหมดและจุดรอที่เกี่ยวข้อง จากนั้นลูป while let จึงจะได้ผ่านจุด await บนการเรียกใช้ recv

ในการได้พฤติกรรมที่เราต้องการ ซึ่งการหน่วงเวลาการนอนหลับเกิดขึ้นระหว่างแต่ละข้อความ เราจำเป็นต้องแยกการทำงานของ tx และ rx ไว้ในบล็อก async ของตนเอง ดังแสดงในโค้ดตัวอย่างที่ 17-11 จากนั้นรันไทม์สามารถประมวลผลแต่ละอันแยกกันได้โดยใช้ trpl::join เช่นเดียวกับในโค้ดตัวอย่างที่ 17-8 อีกครั้ง เรารอคอยผลลัพธ์ของการเรียกใช้ trpl::join ไม่ใช่ฟิวเจอร์แต่ละตัว หากเรารอคอยฟิวเจอร์แต่ละตัวตามลำดับ เราก็จะกลับไปสู่การทำงานตามลำดับเดิมอีกครั้ง—ซึ่งเป็นสิ่งที่เรารพยายาม ไม่ ทำ

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let tx_fut = async {
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("future"),
            ];

            for val in vals {
                tx.send(val).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let rx_fut = async {
            while let Some(value) = rx.recv().await {
                println!("received '{value}'");
            }
        };

        trpl::join(tx_fut, rx_fut).await;
    });
}

ด้วยโค้ดอัปเดตในโค้ดตัวอย่างที่ 17-11 ข้อความจะถูกพิมพ์ที่ช่วงเวลาเว้นห่างกัน 500 มิลลิวินาที แทนที่จะรีบมาพร้อมกันทั้งหมดหลังจาก 2 วินาที

การย้ายความเป็นเจ้าของเข้าไปในบล็อก Async

อย่างไรก็ตาม โปรแกรมนี้ยังคงไม่มีวันจบการทำงาน เนื่องจากวิธีที่ลูป while let โต้ตอบกับ trpl::join:

  • ฟิวเจอร์ที่คืนค่าจาก trpl::join จะทำงานเสร็จสิ้นก็ต่อเมื่อฟิวเจอร์ส ทั้งสอง ที่ส่งเข้าไปทำงานเสร็จสิ้นแล้ว
  • ฟิวเจอร์ tx_fut จะทำงานเสร็จสิ้นเมื่อมันทำงานนอนหลับเสร็จหลังจากส่งข้อความสุดท้ายใน vals
  • ฟิวเจอร์ rx_fut จะไม่เสร็จสิ้นจนกว่าลูป while let จะจบลง
  • ลูป while let จะไม่จบลงจนกว่าการรอคอย rx.recv จะได้ผลลัพธ์เป็น None
  • การรอคอย rx.recv จะคืนค่า None ก็ต่อเมื่ออีกฝั่งหนึ่งของแชนเนลถูกปิดลง
  • แชนเนลจะปิดลงก็ต่อเมื่อเราเรียกใช้ rx.close หรือเมื่อฝั่งส่ง tx ถูกคืนค่า (drop)
  • เราไม่ได้เรียกใช้ rx.close ที่ใดเลย และ tx จะไม่ถูกคืนค่าจนกว่าบล็อก async นอกสุดที่ส่งไปยัง trpl::block_on จะจบลง
  • บล็อกไม่สามารถจบลงได้เพราะมันถูกบล็อกตรงการรัน trpl::join จนเสร็จสิ้น ซึ่งทำให้เราวนกลับไปที่ส่วนบนสุดของรายการนี้อีกครั้ง

ในตอนนี้ บล็อก async ที่เราส่งข้อความเพียงแค่ ยืม tx เพราะการส่งข้อความไม่ได้ต้องการความเป็นเจ้าของ แต่ถ้าเราสามารถ ย้าย (move) tx เข้าไปในบล็อก async นั้นได้ มันจะถูกคืนค่าทันทีเมื่อบล็อกนั้นจบลง ในหัวข้อ “การจับการอ้างอิงหรือการย้ายความเป็นเจ้าของ” ในบทที่ 13 คุณได้เรียนรู้วิธีใช้คีย์เวิร์ด move กับโคลเชอร์ และดังที่อภิปรายในหัวข้อ “การใช้โคลเชอร์แบบ move กับเธรด” ในบทที่ 16 เรามักต้องย้ายข้อมูลเข้าไปในโคลเชอร์เมื่อทำงานกับเธรด กลไกพื้นฐานเดียวกันนี้ก็ประยุกต์ใช้กับบล็อก async ด้วย ดังนั้นคีย์เวิร์ด move จึงทำงานกับบล็อก async เช่นเดียวกับที่ทำกับโคลเชอร์

ในโค้ดตัวอย่างที่ 17-12 เราเปลี่ยนบล็อกที่ใช้ส่งข้อความจาก async เป็น async move

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let tx_fut = async move {
            // --snip--
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("future"),
            ];

            for val in vals {
                tx.send(val).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let rx_fut = async {
            while let Some(value) = rx.recv().await {
                println!("received '{value}'");
            }
        };

        trpl::join(tx_fut, rx_fut).await;
    });
}

เมื่อเรารันเวอร์ชัน นี้ ของโค้ด มันจะปิดตัวลงอย่างนุ่มนวลหลังจากข้อความสุดท้ายถูกส่งและรับแล้ว ถัดไป ลองมาดูกันว่าต้องเปลี่ยนอะไรบ้างหากต้องการส่งข้อมูลจากฟิวเจอร์มากกว่าหนึ่งตัว

การร่วมฟิวเจอร์สจำนวนมากด้วยมาโคร join!

แชนเนล async นี้ยังเป็นแชนเนลแบบผู้ผลิตหลายรายเช่นกัน ดังนั้นเราจึงสามารถเรียกใช้ clone บน tx ได้หากเราต้องการส่งข้อความจากหลายๆ ฟิวเจอร์ส ดังแสดงในโค้ดตัวอย่างที่ 17-13

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let tx1 = tx.clone();
        let tx1_fut = async move {
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("future"),
            ];

            for val in vals {
                tx1.send(val).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let rx_fut = async {
            while let Some(value) = rx.recv().await {
                println!("received '{value}'");
            }
        };

        let tx_fut = async move {
            let vals = vec![
                String::from("more"),
                String::from("messages"),
                String::from("for"),
                String::from("you"),
            ];

            for val in vals {
                tx.send(val).unwrap();
                trpl::sleep(Duration::from_millis(1500)).await;
            }
        };

        trpl::join!(tx1_fut, tx_fut, rx_fut);
    });
}

ขั้นแรก เราโคลน tx สร้างเป็น tx1 ภายนอกบล็อก async แรก เราย้าย tx1 เข้าไปในบล็อกนั้นเหมือนกับที่เราทำก่อนหน้านี้กับ tx จากนั้นในภายหลัง เราย้าย tx ดั้งเดิมเข้าไปในบล็อก async ใหม่ ซึ่งเราส่งข้อความเพิ่มเติมโดยมีการหยุดพักที่ช้าลงเล็กน้อย เราใส่บล็อก async ใหม่นี้ไว้หลังบล็อก async สำหรับรับข้อความ แต่มันสามารถอยู่ก่อนหน้าได้เช่นกัน คีย์สำคัญคือลำดับที่ฟิวเจอร์สถูกรอคอย (await) ไม่ใช่ลำดับที่พวกมันถูกสร้างขึ้น

บล็อก async ทั้งสองสำหรับการส่งข้อความจำเป็นต้องเป็นบล็อก async move เพื่อให้ทั้ง tx และ tx1 ถูกคืนค่าเมื่อบล็อกเหล่านั้นจบลง มิฉะนั้น เราจะจบลงด้วยลูปไม่รู้จบแบบเดิมที่เราเริ่มตั้งต้น

ท้ายที่สุด เราเปลี่ยนจาก trpl::join เป็น trpl::join! เพื่อจัดการกับฟิวเจอร์เพิ่มเติม: มาโคร join! จะรอคอยฟิวเจอร์สจำนวนกี่ตัวก็ได้โดยที่เราทราบจำนวนของฟิวเจอร์สในขั้นตอนคอมไพล์ เราจะอภิปรายเรื่องการรอคอยคอลเลกชันของฟิวเจอร์สที่ไม่ทราบจำนวนในภายหลังของบทนี้

คราวนี้เราจะเห็นข้อความทั้งหมดจากฟิวเจอร์สผู้ส่งทั้งสองตัว และเนื่องจากฟิวเจอร์สผู้ส่งใช้การหยุดพักที่แตกต่างกันเล็กน้อยหลังจากส่ง ข้อความจึงถูกรับในช่วงเวลาเว้นห่างที่แตกต่างกันเหล่านั้นด้วย:

received 'hi'
received 'more'
received 'from'
received 'the'
received 'messages'
received 'future'
received 'for'
received 'you'

เราได้สำรวจวิธีใช้การส่งข้อความเพื่อส่งข้อมูลระหว่างฟิวเจอร์ส วิธีที่โค้ดภายในบล็อก async รันตามลำดับ วิธีการย้ายความเป็นเจ้าของเข้าไปในบล็อก async และวิธีการร่วมฟิวเจอร์สหลายตัว ถัดไป ลองมาอภิปรายถึงวิธีและเหตุผลในการบอกรันไทม์ว่ามันสามารถสลับไปทำแทสก์อื่นได้