管道 · Kotin知識點難點歸納總結

## 管道 ### 管道簡介當通過async()啟動一個協程時，會返回一個defferd類型的對象，defferd相當于是輸出一個具體的值，此時也可以通過管道（Channel）的方式來輸出具體的值。管道提供了一種傳輸的價值流。管道在概念上與BlockingQueue（阻塞隊列）非常類似，但是區別在于管道不是一個阻塞put操作而是一個暫停發送操作，不是一個阻塞take操作而是一個暫停接收操作。阻塞隊列會阻塞線程，而管道則不會阻塞線程。管道中有兩個方法，分別是send()和receive()，這兩個方法分別用于發送數據和接收數據。接下來我們通過一個案例來演示管道的send()和receive()方法，具體代碼如下所示。 ``` import kotlinx.coroutines.experimental.channels.Channel import kotlinx.coroutines.experimental.delay import kotlinx.coroutines.experimental.launch import kotlinx.coroutines.experimental.runBlocking fun main(args: Array<String>): Unit = runBlocking { val channel = Channel<Int>() launch { (1..4).forEach { channel.send(it * 10) delay(1000L) } } repeat(4) { val result = channel.receive() print("result=${result}\t") } } ``` 運行結果： ``` result=10 result=20 result=30 result=40 ``` 上述代碼中，Channel表示管道，在forEach循環中每間隔一秒會執行一次channel中的send()方法，來發送協程中的數據信息，接著通過repeat()方法來循環執行接收數據的代碼，該方法中的參數4表示接收的信息數量，在repeat()方法中通過channel中的receive()方法來接收發送的數據信息并打印出來。 ### 管道的關閉管道與阻塞隊列比較類似。但是管道與阻塞隊列的區別是，第一，隊列是阻塞的，管道是非阻塞的；第二，管道可以通過close()方法進行關閉，當沒有更多數據需要添加到管道中時，管道就可以進行關閉。在管道的接收端，通常使用for循環接收管道發送的數據，從概念上來講，結束就像發送了一個特殊的密碼令牌給該頻道，一旦接收到這個關閉標記，迭代就會停止，之后所有發送的數據不會被接收。接下來我們通過一個案例來演示關閉管道后數據的接收情況，具體代碼如下所示。 ``` import kotlinx.coroutines.experimental.channels.Channel import kotlinx.coroutines.experimental.delay import kotlinx.coroutines.experimental.launch import kotlinx.coroutines.experimental.runBlocking fun main(args: Array<String>): Unit = runBlocking { val channel = Channel<Int>() launch { (1..3).forEach { channel.send(it * 10) println("發送端的關閉狀態=${channel.isClosedForSend}") delay(1000L) } //關閉管道 channel.close() println("管道關閉后發送端的關閉狀態=${channel.isClosedForSend}") println("管道關閉后接收端的關閉狀態=${channel.isClosedForReceive}") } repeat(10) { val result = channel.receive() println("result=$result 接收端的關閉狀態=${channel. isClosedForReceive}") } } ``` 運行結果： ``` 發送端的關閉狀態=false result=10接收端的關閉狀態=false 發送端的關閉狀態=false result=20接收端的關閉狀態=false 發送端的關閉狀態=false result=30接收端的關閉狀態=false 管道關閉后發送端的關閉狀態=true 管道關閉后接收端的關閉狀態=true Exception in thread"main"kotlinx.coroutines.experimental.channels. ClosedReceiveChannelException:Channel was closed ``` 根據上述代碼的運行結果可知，管道分為發送端關閉狀態和接收端關閉狀態，當管道被關閉之前，發送端的關閉狀態都為false，接收端的關閉狀態也為false，等接收完所有元素之后管道關閉，此時發送端的關閉狀態與接收端的關閉狀態才為true。管道被關閉之后就無法接收其他數據，否則，程序就會報錯。由于上述代碼中repeat()方法中傳遞的參數為10，也就是當接收完管道發送過來的3個數據之后，在接收端還在循環進行接收數據，此時程序運行結果就報錯了。 ### 生產者與消費者協程產生一系列元素的模式比較普遍，這通常是在并發代碼中發現的“生產者—消費者”模式的一部分，可以將生產者抽象為一個以通道為參數的函數，但是生產者的結果必須是從函數中返回的。接下來，通過管道來生成一個生產者和消費者模式，在管道中可以通過produce來生成一個協程，在協程中通過管道來發送一些數據。接下來我們通過一個案例來演示通過管道來生成生產者與消費者，具體代碼如下所示。 ``` import kotlinx.coroutines.experimental.channels.consumeEach import kotlinx.coroutines.experimental.channels.produce import kotlinx.coroutines.experimental.delay import kotlinx.coroutines.experimental.runBlocking //生產者 fun produceSquares() = produce<Int> { (1..5).forEach { send(it * 10) delay(1000L) } } //消費者 suspend fun consumeSquares() { val squares = produceSquares() //接收生產者發送的信息 squares.consumeEach { //類似于for循環 print("it=$it \t") } } fun main(args: Array<String>): Unit = runBlocking { consumeSquares() } ``` 運行結果： ``` it=10　it=20　it=30　it=40　it=50 ``` 上述代碼中，通過produce生成了一個協程。在這個協程中通過管道中的send()方法將信息發送出去，這個協程的返回值是一個函數produceSquares()，這個函數是一個生產者。接著在consumeSquares()方法中獲取生產者發送的信息，并通過擴展函數consumeEach()可以替代for循環。管道的發送順序和接收順序是一致的，管道主要用于線程間通信和父子進程間的通信。如果以后遇到需要描述生產者和消費者的模式時，可以通過管道來進行演示。 ### 管道緩存區當默認創建一個管道時，這個管道是沒有緩沖區的，發送端和接收端彼此間相遇時才可以進行發送和接收的操作，也就是說當發送端發送時，首先調用send()方法，發送完信息之后就必須要接收；接收端接收完之后才可以發送下一條信息。如果發送完信息之后還沒有被接收，此時程序就暫時停在這個地方，并不屬于阻塞，屬于掛起，等到后續接收完信息之后才會繼續發送信息。接收端接收時也是一樣的，如果接收時沒有數據發送過來，此時程序就暫時停止，直到有信息發送過來才會進行接收，在這里的等待也屬于掛起而不是阻塞。為了解決這個暫時停止的問題，可以在程序中創建管道的緩沖區。在創建緩沖通道時，可以設置緩沖區的大小，其中Channel()函數中傳遞的capacity參數是來指定緩沖區大小的，緩沖區允許發送者在掛起之前發送多個元素，類似于BlockingQueue指定的容量，當緩沖區沒有滿時，無論發送端發送的信息有沒有被接收都可以一直向緩沖區存放發送的元素，直到緩沖區被存放滿時，程序才會掛起，掛起之后等待后續接收這些信息，接收完之后才會繼續進行這樣的操作。接下來我們通過一個案例來演示管道的緩沖區，具體代碼如下所示。 ``` import kotlinx.coroutines.experimental.channels.Channel import kotlinx.coroutines.experimental.delay import kotlinx.coroutines.experimental.launch import kotlinx.coroutines.experimental.runBlocking fun main(args: Array<String>): Unit = runBlocking { val channel = Channel<Int>(3) //創建緩沖通道 val sender = launch(coroutineContext) { //啟動協程 repeat(10) { println("sending $it") //打印發送的每個元素 channel.send(it) //發送元素，當緩沖區已滿時將暫停發送 } } delay(1000L) sender.cancel() //取消協程sender println("") } ``` 運行結果： ``` sending 0 sending 1 sending 2 sending 3 ``` 上述代碼中，通過Channel創建了一個緩沖區，這個緩沖區的大小設置為3，也就是可以向緩沖區中存放3個元素。當發送完前3個元素之后，這3個元素已經緩存在管道中，等著接收端進行接收。當發送第4個元素時，發現這個緩沖區已經沒有空間了，這個發送此時就變為掛起狀態。由于第15行代碼中的cancel()方法的返回值是布爾類型的，而main()函數的返回值是Unit，也就是沒有返回值，因此需要在調用cancel()方法的下方隨意添加一行沒有返回值的代碼，不然程序會報錯。 ## 本章小結本章主要介紹了Kotlin中的協程，詳細介紹了協程的概念、協程的取消以及管道。通過對本章的學習，讀者可以掌握Kotlin程序中協程的使用方法。要求讀者必須掌握本章內容，便于后續開發Kotlin程序。【思考題】 1. 請思考線程與協程的效率對比。 2. 請思考協程是如何取消的。