Channel: คืออะไร ประเภทของช่องสัญญาณและโคโรทีนใน Kotlin

Channel ใน Kotlin คืออะไร?

Channel มีแนวคิดคล้ายกับ BlockingQueue จาก Java แต่ใช้ฟังก์ชัน suspend send() และ receive() แทน put() และ take() แบบบล็อก นักพัฒนา Kotlin ใช้ Channel เพื่อจัดระเบียบการแลกเปลี่ยนข้อมูลระหว่างโคโรทีนโดยไม่ต้องซิงโครไนซ์ผ่านหน่วยความจำที่ใช้ร่วมกัน ช่องสัญญาณรับประกันการส่งตามลำดับ — ลำดับการส่งตรงกับลำดับการรับ

การสร้างช่องสัญญาณ

ในการสร้าง Channel จะเรียกฟังก์ชันโรงงาน Channel<T>(capacity) พารามิเตอร์ capacity กำหนดประเภทของช่องสัญญาณ: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) หรือตัวเลขเฉพาะ ประเภทขององค์ประกอบ T ถูกระบุผ่านเจเนอริก การปิดช่องสัญญาณด้วย close() ส่งสัญญาณว่าจะไม่มีองค์ประกอบใหม่มา

Send และ Receive

send(value) เป็นฟังก์ชัน suspend ที่ระงับโคโรทีนผู้ส่งหากช่องสัญญาณเต็ม receive() เป็นฟังก์ชัน suspend ที่ระงับผู้รับหากช่องสัญญาณว่าง ทางเลือก trySend() และ tryReceive() เป็นเวอร์ชันที่ไม่บล็อกซึ่งส่งคืน Boolean หรือ null เมื่อการดำเนินการไม่สามารถทำได้ มีประโยชน์ในบริบทที่ไม่ใช่ suspend

ประเภทของ Channel ใน kotlinx.coroutines

Kotlin มี Channel สี่แบบผ่านความจุของบัฟเฟอร์: Rendezvous (ความจุ 0), Buffered (ความจุ N), Conflated (ความจุ 1, เขียนทับ) และ Unlimited (ความจุ Int.MAX_VALUE) แต่ละประเภทแก้ปัญหาของตัวเอง ตั้งแต่การซิงโครไนซ์ที่เข้มงวดไปจนถึงการบัฟเฟอร์ข้อมูลจำนวนมาก

Rendezvous Channel เป็นแบบที่เข้มงวดที่สุด: send() จะถูกบล็อกจนกว่า receive() จะถูกเรียกในโคโรทีนอื่น โดยพื้นฐานแล้ว มันคือจุดนัดพบของสองโคโรทีน เหมาะสำหรับการจับมือที่เข้มงวดเมื่อผู้ส่งต้องรอให้ผู้รับประมวลผลองค์ประกอบ การสูญเสียข้อมูลถูกตัดออก — send จะไม่เสร็จสมบูรณ์จนกว่า receive จะถูกดำเนินการ

Conflated Channel เก็บเฉพาะค่าสุดท้ายที่ส่ง หากผู้ส่งวางองค์ประกอบใหม่ก่อนที่ผู้รับจะรับค่าเก่า ค่าเก่าจะถูกทิ้ง Conflated Channel มีประโยชน์สำหรับสถานะ UI: หากผู้ใช้เปลี่ยนแถบเลื่อนอย่างรวดเร็ว ค่ากลางสามารถทิ้งได้และประมวลผลเฉพาะค่าสุดท้าย

Producer-Consumer ด้วยช่องสัญญาณ

รูปแบบ Producer-Consumer แบบคลาสสิกบน Channel ถูกนำไปใช้ผ่านโคโรทีนแบบขนาน Producer เรียก send(value) ในลูป consumer เรียก receive(value) ผู้ผลิตและผู้บริโภคสามารถทำงานบน Dispatchers ที่แตกต่างกัน: ผู้ผลิตบน Dispatchers.IO ผู้บริโภคบน Dispatchers.Main Channel ซิงโครไนซ์การเข้าถึงโดยอัตโนมัติโดยไม่ต้องใช้ Lock หรือ synchronized

Fan-out — ผู้บริโภคหลายรายบนช่องสัญญาณเดียว แต่ละองค์ประกอบไปยังผู้บริโภคหนึ่งรายเท่านั้น (การกระจายแบบ round-robin) Fan-in — ผู้ผลิตหลายรายเขียนลงในช่องสัญญาณเดียว โคโรทีนผู้ส่งแข่งขันกันเพื่อส่ง แต่ลำดับขององค์ประกอบจะถูกรักษาไว้ ทั้งสองสถานการณ์ไม่ต้องการการซิงโครไนซ์เพิ่มเติม

Produce เป็นตัวสร้างโคโรทีนที่สร้างช่องสัญญาณที่มีการปิดอัตโนมัติ ฟังก์ชัน produce { } ส่งคืน ReceiveChannel — ช่องสัญญาณแบบอ่านอย่างเดียวสำหรับผู้บริโภค ภายในตัวสร้าง send() จะส่งข้อมูล และเมื่อบล็อกเสร็จสมบูรณ์หรือมีข้อยกเว้น ช่องสัญญาณจะปิดโดยอัตโนมัติ ป้องกันการรั่วไหล

Select และการมัลติเพล็กซ์

ไลบรารี kotlinx.coroutines มี select — นิพจน์ที่รอช่องสัญญาณแรกที่เสร็จสมบูรณ์จากหลายทางเลือก Select อนุญาตให้มัลติเพล็กซ์หลายช่องสัญญาณ: ตัวอย่างเช่น รอข้อมูลจากสองแหล่งและประมวลผลแหล่งที่ตอบกลับก่อน ไวยากรณ์ — select<T> { channel1.onReceive { } channel2.onReceive { } } นี่เป็นทางเลือกสำหรับตัวดำเนินการ amb ใน Rx

ตัวอย่างโค้ด Channel

ตัวอย่างแรกคือ Rendezvous Channel อย่างง่ายที่ผู้ส่งรอการรับ:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("ส่งแล้ว")
}

scope.launch {
    val msg = channel.receive()
    println("ได้รับ: $msg")
}

ตัวอย่างที่สอง — ผู้บริโภคหลายรายบนช่องสัญญาณเดียว (fan-out):

val channel = Channel<Int>(Channel.UNLIMITED)

scope.launch {
    for (x in 1..10) channel.send(x)
    channel.close()
}

repeat(2) { id ->
    scope.launch {
        for (msg in channel) {
            println("Consumer #$id: $msg")
        }
    }
}

ตัวอย่างที่สาม — การใช้ตัวสร้าง produce พร้อมการจัดการข้อผิดพลาด:

val source = produce {
    for (i in 1..5) {
        delay(200)
        send(i)
    }
}

scope.launch {
    source
        .consumeAsFlow()
        .catch { println("ข้อผิดพลาด: $it") }
        .collect { println("องค์ประกอบ: $it") }
}

Channel vs Flow

Channel เป็นพรีมิทีฟร้อน: ข้อมูลถูกปล่อยออกโดยไม่ขึ้นกับสมาชิก Flow เป็นเย็น: ข้อมูลถูกสร้างขึ้นเมื่อสมัครสมาชิก Channel รองรับผู้ผลิตและผู้บริโภคหลายรายพร้อมการส่งแต่ละองค์ประกอบไปยังผู้บริโภคหนึ่งราย (fan-out) Flow ไม่ได้ออกแบบมาสำหรับผู้ผลิตอิสระหลายราย

Channel ใช้บัฟเฟอร์ที่มีความจุที่กำหนดค่าได้และฟังก์ชัน suspend send/receive สำหรับการจัดการแรงดันย้อนกลับ Flow ใช้กลไก suspend collect พร้อมแรงดันย้อนกลับอัตโนมัติผ่านโคโรทีน Channel เป็นเครื่องมือระดับต่ำสำหรับสถานการณ์เฉพาะ: การแปลง callback, โมเดลแอคเตอร์, คิวงานที่มีผู้ส่งหลายราย

สำหรับสถานการณ์ประจำวันใน Android (สถานะ UI, สตรีมเชิงปฏิกิริยาจากฐานข้อมูล) Google แนะนำ Flow มากกว่า Channel ควรใช้ Channel เมื่อต้องการการแลกเปลี่ยนข้อมูลร้อนระหว่างโคโรทีนพร้อมการควบคุมบัฟเฟอร์ที่แม่นยำ หรือเมื่อแปลงอินเทอร์เฟซ callback ผ่าน callbackFlow ซึ่งการใช้งานภายในใช้ Channel

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

การจัดการวงจรชีวิตของช่องสัญญาณเป็นส่วนสำคัญของการทำงานกับ Channel ช่องสัญญาณต้องถูกปิดเมื่อส่งข้อมูลทั้งหมดแล้วเพื่อให้ผู้บริโภคสามารถทำการวนซ้ำให้เสร็จสมบูรณ์ การเรียก channel.close() ส่งสัญญาณว่าจะไม่มีองค์ประกอบใหม่มา ผู้บริโภคสามารถวนซ้ำผ่าน for (item in channel) — ลูปจะสิ้นสุดโดยอัตโนมัติหลังจาก close() และบัฟเฟอร์หมด หรือผู้บริโภคสามารถเรียก receive() ในลูปพร้อมการจัดการ ClosedReceiveChannelException

Channel ถูกใช้อย่างแข็งขันใน Android สำหรับการนำ EventBus ไปใช้โดยไม่ต้องพึ่งพา: Channel<Event> ทั่วโลกที่มีกลยุทธ์ Broadcast อนุญาตให้ส่งเหตุการณ์จากจุดใดก็ได้ในแอปพลิเคชัน ซึ่งแตกต่างจากบัสที่ใช้ LiveData Channel ไม่ผูกกับวงจรชีวิตและไม่ต้องการการล้างเมื่อเปลี่ยนระหว่างหน้าจอ send() จาก ViewModel และ receive() ใน Activity/Fragment ผ่าน lifecycleScope ให้การสื่อสารที่ปลอดภัยชนิดโดยไม่ต้องมีคลาส Event ผู้บริโภคหลายรายบน Channel กระจายโหลด — แต่ละองค์ประกอบถูกประมวลผลหนึ่งครั้ง ซึ่งป้องกันการประมวลผลซ้ำของเหตุการณ์เดียวในสมาชิกที่แตกต่างกัน

ในระบบแอคเตอร์ Channel ทำหน้าที่เป็นพื้นฐานสำหรับการนำกล่องจดหมาย — คิวข้อความสำหรับแอคเตอร์ — ไปใช้ แอคเตอร์คือโคโรทีนที่อ่านข้อความจาก Channel ในลูปและประมวลผลตามลำดับ วิธีนี้รับประกันว่าข้อความแต่ละข้อความถูกประมวลผลตามลำดับการส่ง โดยไม่มีการแข่งขันข้อมูล Kotlin ไม่มีแอคเตอร์ในตัวเป็นชนิด (ต่างจาก Akka) แต่ Channel + launch เป็นตัวแทนที่มีน้ำหนักเบา

สำหรับการแลกเปลี่ยนแบบสองทิศทาง จะใช้ คู่ช่องสัญญาณ: หนึ่งช่องสำหรับคำขอจากไคลเอ็นต์ไปยังเซิร์ฟเวอร์ สองสำหรับการตอบกลับจากเซิร์ฟเวอร์ไปยังไคลเอ็นต์ ตัวอย่างเช่น เมื่อนำ Pipe ไปใช้ในแอปพลิเคชันแบบหลายเธรด: ผู้ผลิตเขียนไปยัง OutputChannel ผู้บริโภคอ่านจาก InputChannel ฟังก์ชัน suspend send และ receive รับประกันว่า Producer-Consumer จะไม่ล้นสแต็กการเรียก เนื่องจากโคโรทีนถูกระงับแทนที่จะถูกบล็อก Channel ที่มีความจุ BUFFERED เหมาะสำหรับสถานการณ์ส่วนใหญ่ที่ความเร็วของผู้ผลิตและผู้บริโภคเท่ากันโดยประมาณ สำหรับสถานการณ์ที่ไม่สมดุล ให้ใช้ UNLIMITED เพื่อให้ผู้ผลิตไม่ถูกระงับเมื่อผู้บริโภคไม่ว่าง — ซึ่งลดความเสี่ยงของการตายร่วมแต่เพิ่มการใช้หน่วยความจำ

การเลือกความจุของ Channel

เมื่อออกแบบสถาปัตยกรรมด้วยช่องสัญญาณ สิ่งสำคัญคือต้องจำ capacity: การเลือกความจุส่งผลโดยตรงต่อพฤติกรรมภายใต้โหลดสูงสุด ช่องสัญญาณที่มีความจุ BUFFERED(N) ทำหน้าที่เป็นบัฟเฟอร์ปรับเรียบ: หากผู้บริโภคช้ากว่าผู้ผลิตชั่วคราว องค์ประกอบจะสะสม หากความเร็วเฉลี่ยของผู้บริโภคต่ำกว่าผู้ผลิตอย่างต่อเนื่อง บัฟเฟอร์จะเต็มและโคโรทีนผู้ส่งจะถูกระงับ — นี่คือแรงดันย้อนกลับอัตโนมัติที่ป้องกันหน่วยความจำล้น

สำหรับการตรวจสอบและการดีบัก Channel ให้ใช้ kotlinx-coroutines-debug: ยูทิลิตี้แสดงจำนวนโคโรทีนที่ทำงานอยู่ สถานะช่องสัญญาณของพวกมัน (เปิด/ปิด จำนวนองค์ประกอบในบัฟเฟอร์) และสแต็กการเรียกของการดำเนินการ send/receive ที่ถูกระงับ Channel ยังสามารถห่อในพร็อกซีบันทึก: คลาส LoggingChannel<T> มอบหมายการเรียกไปยัง Channel จริง บันทึกการดำเนินการ send, receive และ close ซึ่งช่วยระบุการรั่วไหลของช่องสัญญาณเมื่อ close() ไม่ถูกเรียกและโคโรทีนผู้บริโภครอองค์ประกอบใหม่อย่างไม่มีที่สิ้นสุด