Heat flow —- pipeline
What is a pipe?
It’s essentially a queue
A channel for transferring data from coroutine to coroutine, used to transfer variables
Kotlin’s channels are similar to JUC’s channels
Why pipes?
Send and receive variables in a way that does not block the current thread with the coroutine
We can think of Kotlin’s channel as BlockingQueue, but instead of blocking the queue put, a channel uses the suspend function send, and instead of blocking the queue take, the suspend function Receive
This gives a channel the advantage of not blocking the current thread
A channel is a data structure that allows one-way transmission of information, from the writing end of the pipe to the reading end of the pipe, all of which are serial and its order is unchanged
How do the pipes work?
Pipeline hello world
fun main(a): Unit = runBlocking {
val channel = Channel<Int>()
launch {
for (x in 1.. 5) {
channel.send(x)
}
}
repeat(5) {
log(channel.receive())
}
log("Done")}Copy the codeSend is suspended until the other party calls Recevie. Receive is suspended if the other party does not send any variables
The function of flow is to transfer variables between streams. What’s the difference? In fact, flow is cold, but channel is hot. Each variable sent can receive a variable immediately
How do pipes iterate and close?
Pipeline iteration
fun main(a): Unit = runBlocking {
val channel = Channel<Int>()
launch {
for (x in 1.. 5) {
channel.send(x * x)
}
}
for (y in channel) {
log(y)
}
log("Done")}Copy the codeThe closing of the pipeline and the judgment of its closing state
@ExperimentalCoroutinesApi
fun main(a): Unit = runBlocking {
val channel = Channel<Int>()
launch {
for (x in 1.. 5) {
if (channel.isClosedForSend) {
log("send: close send:${channel.isClosedForSend}, receive: ${channel.isClosedForReceive}")
break
}
channel.send(x)
}
}
repeat(5) {
if (it >= 3) {
channel.close()
}
if (channel.isClosedForReceive) {
log("receive: close send:${channel.isClosedForSend}, receive: ${channel.isClosedForReceive}")
return@repeat
}
log(channel.receive())
}
log("Done")}Copy the codeProducer and consumer models of pipelines
@ExperimentalCoroutinesApi
fun main(a): Unit = runBlocking {
val receiver = produce {
for (x in 1.. 5) {
send(x * x)
}
}
receiver.consumeEach { log(it) }
log("Done")}Copy the codeThe underlying function produce will start a new coroutine
The result of one pipe can be given to another pipe
fun CoroutineScope.produceNumber(a) = produce<Int> {
repeat(10) {
send(it)
}
}
fun CoroutineScope.square(number: ReceiveChannel<Int>) = produce<Int> {
for (i in number) {
send(i * i)
}
}
fun main(a): Unit = runBlocking {
val number = produceNumber()
val receiver = square(number)
repeat(10) {
log(receiver.receive())
}
log("Done")
coroutineContext.cancelChildren()
}
Copy the codePipes and infinite primes
@ExperimentalCoroutinesApi
fun CoroutineScope.numbersProducer(start: Int) = produce {
var n = start
while (true) send(n++)
}
@ExperimentalCoroutinesApi
fun CoroutineScope.filterPrimes(numbers: ReceiveChannel<Int>, prime: Int) =
produce {
for (x in numbers) {
if(x % prime ! =0) {
send(x)
}
}
}
@ExperimentalCoroutinesApi
fun main(a): Unit = runBlocking {
var numbers = numbersProducer(2)
while (true) {
val prime = numbers.receive()
log("$prime")
numbers = filterPrimes(numbers, prime)
delay(1000)}}Copy the codeChannel buffer
We can set a buffer for the pipe, otherwise the pipe can only put one element at a time
fun main(a): Unit = runBlocking {
val channel: Channel<Int> = Channel<Int>(capacity = 2)
launch {
repeat(5) {
log("Sending $it")
channel.send(ThreadLocalRandom.current().nextInt(0.10))
}
}
launch {
repeat(5) {
log("Reading: ${channel.receive()}")}}}Copy the codeAnd then you’ll notice that every time you pass three elements into the pipe, and then you read, yeah remember that the pipe buffer is limited to two, but then you can pass three elements into the pipe, so be careful here
[main] Sending 0
[main] Sending 1
[main] Sending 2
[main] Reading: 3
[main] Reading: 4
[main] Reading: 6
[main] Sending 3
[main] Sending 4
[main] Reading: 8
[main] Reading: 7
Copy the codeFan out: One more round
A message is distributed to multiple coroutines
@ExperimentalCoroutinesApi
private fun CoroutineScope.produceNumber(a) = produce {
var x = 1
while (true) {
send(x++)
delay(100)}}private fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
for (msg in channel) {
log("id = $id , received: $msg")}}@ExperimentalCoroutinesApi
fun main(a): Unit = runBlocking {
val product = produceNumber()
repeat(5) {
launchProcessor(it, product)
}
}
Copy the codeThe console prints out different coroutines that are accepting
[main @coroutine#3] id = 0 , received: 1
[main @coroutine#3] id = 0 , received: 2
[main @coroutine#4] id = 1 , received: 3
[main @coroutine#5] id = 2 , received: 4
[main @coroutine#6] id = 3 , received: 5
[main @coroutine#7] id = 4 , received: 6
[main @coroutine#3] id = 0 , received: 7
[main @coroutine#4] id = 1 , received: 8
[main @coroutine#5] id = 2 , received: 9
[main @coroutine#6] id = 3 , received: 10
Copy the codeThe radio channel
The previous fanning-out scenario is a multi-path scenario, but there is a problem. One coroutine sends 1-5 numbers to multiple coroutines. If it is the scenario above, only one coroutine can get the number 1, and the other coroutines cannot get the number 1, and the same is true between 2 and 5
The scenario is that the coroutine A sends 1 to 5 numbers to the five coroutines B, C, D, E, and F, and each thread needs to receive 1 to 5 numbers
@ObsoleteCoroutinesApi
fun main(a): Unit = runBlocking {
val channel = BroadcastChannel<Int>(Channel.BUFFERED)
launch {
List(5) {
delay(200)
channel.send(it)
}
channel.close()
}
List(5) {
launch {
val receiveChannel = channel.openSubscription()
for (i in receiveChannel) {
log("received: $i")
}
}
}.joinAll()
}
Copy the codeFan in: multiple one
suspend fun sendString(channel: Channel<String>, s: String, time: Long) {
while (true) {
delay(time)
channel.send(s)
}
}
fun main(a): Unit = runBlocking {
val channel = Channel<String>()
launch { sendString(channel, "foo".200L) }
launch { sendString(channel, "BAR".500L) }
repeat(6) {
log(channel.receive())
}
coroutineContext.cancelChildren()
}
Copy the codePipelines are fair: first in, first out
The sending and receiving channels follow the first-in, first-out principle
data class Ball(var hits: Int)
suspend fun player(name: String, channel: Channel<Ball>) {
for (ball in channel) {
ball.hits++
log("$name.${ball.hits}")
delay(100)
channel.send(ball)
}
}
fun main(a): Unit = runBlocking {
val channel = Channel<Ball>()
launch { player("ping", channel) }
launch { player("pong", channel) }
channel.send(Ball(0))
delay(1000)
// Terminate the loop
coroutineContext.cancelChildren()
}
Copy the codeSelect multiplexing
Select multiplexing is commonly heard in many places, such as Linux select mechanism multiplexing, in select if we are wrong to use a thread to monitor multiple file handles (I remember limit is 1024?). , similar to a thread constantly polling each array for a value to determine if there is an event that needs to be processed
What is multiplexing?
The multiplexing of select in Kotlin looks like this:
Select the available and fastest function to execute from the onXXX events of multiple suspended functions
Look at the following code:
@ExperimentalCoroutinesApi
fun CoroutineScope.fizz(a) = produce {
while (true) {
delay(300)
send("Fizz")}}@ExperimentalCoroutinesApi
fun CoroutineScope.buzz(a) = produce {
while (true) {
delay(500)
send("Buzz")}}suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
select<Unit> {
fizz.onReceive { s -> log("fizz -> $s") }
buzz.onReceive { s -> log("buzz -> $s")}}}@ExperimentalCoroutinesApi
fun main(a): Unit = runBlocking {
val fizz = fizz()
val buzz = buzz()
selectFizzBuzz(fizz, buzz)
// fizz.cancel()
// buzz.cancel()
coroutineContext.cancelChildren()
}
Copy the code[main] fizz -> Fizz
Here we have an obe
fun CoroutineScope.fizz(a) = produce {
while (true) {
delay(300)
send("Fizz")}}Copy the codeIf we just use CoroutineScope, we can just use produce instead of having to figure out how to build a CoroutineScope CoroutineScope and then use produce, right
When I was writing the code, I wanted to use withContext, but it didn’t work, and I looked at it and realized that the thread was blocked at send
private suspend fun fizz(a) = withContext(Dispatchers.IO) {
log("fizz")
produce {
repeat(5) {
delay(300)
send("Fizz") // Block here}}}Copy the codeAnd then because it’s blocked at send and the withContext itself has to block and wait for its internal coroutine to complete, but send is blocked again, the internal code can’t complete, so it’s blocked indefinitely, The following code blocks on val fizz1 = fizz1() and cannot run further
@Test
fun test03(a) = runBlocking {
val fizz = fizz()
log("fizz: $fizz")}Copy the codeSomething like this:
@Test
fun test03(a) = runBlocking {
coroutineScope {
// Fizz is blocked, infinitely blocked, and coroutineScope says,
If the internal coroutine does not end, the main thread is constrained (but not blocked),
// Do not allow the main thread to run out of scope, that is, cannot execute
// log("fizz1: $fizz1"
val fizz = fizz()
}
log("fizz1: $fizz1")}Copy the codeRemember to analyze the above code in detail, especially comments
abnormal
Cancelling coroutines throws a CancellationException at the point of suspension, which is normally invisible and requires special catching
Suspend point: The point where the suspend function + asynchronous operation is called the suspend point
suspend fun main(a) {
coroutineScope {
val job = launch {
try {
repeat(100000) {
delay(1000)
println("${Thread.currentThread()}: ping")}}catch (e: Exception) {
e.printStackTrace()
}
}
delay(3000)
job.cancel()
}
}
Copy the codeThread[DefaultDispatcher-worker-1.5,main]: ping
Thread[DefaultDispatcher-worker-1.5,main]: ping
kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled; job=StandaloneCoroutine{Cancelling} @ 5d5390a2
Copy the codeAutomatically propagates exceptions and exposes exceptions to users
There are two kinds of coroutine constructors: Launch exceptions (and actor) or expose exceptions (async and produce) to users. If the coroutine constructor is created in a root coroutine (that is, no coroutine child is it), the exception propagated automatically will be treated as an uncaught exception and will be thrown directly. Exposing exceptions to the user depends on the user ultimately consuming the exception (await or receive)
The root coroutine that automatically propagates the exception throws it directly
Look at the code
suspend fun main(a) {
coroutineScope {
val job = GlobalScope.launch {
throw IndexOutOfBoundsException() // An error is reported here
}
job.join()
val deferred = GlobalScope.async {
throw ArithmeticException()
}
try {
deferred.await() // An error is reported here
}
catch (e: Exception) {
log("I caught an exception here,$e")}}}Copy the codeIf we no matter, the two will be an error, but the location of the error is different, launch the location of the error is throw IndexOutOfBoundsException () and async mode error is deferred. Await () this position
Exceptions that are not root coroutines are always propagated layer by layer to the root coroutine
Propagation of exceptions also involves interscope issues, as the GlobalScope we used earlier to create coroutines means that the exception is in a separate coroutineScope, which is not the case with the coroutine we created with coroutineScope
If a child coroutine encounters an exception, it throws an exception to its parent coroutine. The parent coroutine, if it has a parent coroutine, continues to pass it to its parent coroutine until it is passed to the root coroutine
So if an exception occurs on any of the child coroutines (with the exception of CancellationException), all the other coroutines will be stopped and no exceptions will occur on any of the other coroutines
The coroutines GlobalScope creates are independent of the parent coroutine of the coroutine in which it is created, which is the root coroutine, so this problem does not occur
It’s going to be a case where the whole coroutine tree is going to end up, and it’s going to be the container that we’re dealing with
It is designed to prevent the whole coroutine tree from failing due to the abnormal propagation of subroutines
fun main(a): Unit = runBlocking {
val scope = CoroutineScope(SupervisorJob())
val job1 = scope.launch {
delay(1000)
throw RuntimeException()
}
val job2 = scope.launch {
repeat(10) {
delay(150)
log("Coroutine 2 keeps running.")
}
}
joinAll(job1, job2)
}
Copy the codeIf job1 fails, joB2 can be executed until the execution is complete
If we need to cancel all coroutines we can call scope.cancel()
supervisorScope
We could also use it like this
suspend fun main(a): Unit = supervisorScope {
launch {
repeat(10) {
delay(200)
log(Subcoroutine alive)
}
}
launch {
delay(1000)
throw AssertionError()
}
}
Copy the codeCan achieve the same effect
But there is a problem with this approach
If a scoped exception exists within the scope, as follows:
suspend fun main(a): Unit = supervisorScope {
launch {
repeat(10) {
delay(200)
log(Subcoroutine alive)
}
}
delay(1000)
throw AssertionError()
}
Copy the codeThe coroutine is no longer allowed to run 10 rounds, and it is handled by an AssertionError that exits the entire container scope
CoroutineExceptionHandler: exception
CoroutineExceptionHandler only to capture the exception thrown directly coroutines (launch this not async)
It is only captured in the context of the CoroutineScope or in the root coroutine
private val handler = CoroutineExceptionHandler { _, throwable ->
log("1 handler $throwable")}@DelicateCoroutinesApi
fun main(a): Unit = runBlocking {
val job = GlobalScope.launch(handler) {
throw AssertionError() // Only this exception will be intercepted
}
val deferred = GlobalScope.async(handler) {
throw ArithmeticException() // This line of code inside async does not catch exceptions
}
joinAll(job, deferred)
}
Copy the codefun main(a) {
val scope = CoroutineScope(handler)
scope.launch {
val job = launch {
throw AssertionError() // Only this exception will be intercepted
}
val deferred = async {
throw ArithmeticException() // This line of code inside async does not catch exceptions
}
joinAll(job, deferred)
}
TimeUnit.SECONDS.sleep(2)}Copy the codeBut there’s a catch:
suspend fun main(a) {
val scope = CoroutineScope(Job())
val job = scope.launch {
launch(handler) {
throw IllegalArgumentException()
}
}
job.join()
}
Copy the codeThe above code does not catch exceptions, as stated earlier “and only in the context of the CoroutineScope or the root coroutine.”
This is inconvenient, however, because you need to add exception handlers to each coroutine
useMETA-INFThe configuration file implements global exceptionsTo view
He doesn’t catch exceptions, he just sees what exceptions are there, and he logs the crash, right
val scope = CoroutineScope(handler)
GlobalScope.launch(handler)
So we can also do this:
class GlobalCoroutinesException : CoroutineExceptionHandler {
override val key: CoroutineContext.Key<*>
get() = CoroutineExceptionHandler
override fun handleException(context: CoroutineContext, exception: Throwable) {
log("Coroutine exception: $exception")}}@DelicateCoroutinesApi
fun main(a): Unit = runBlocking {
val job = GlobalScope.launch {
throw AssertionError()
}
val deferred = GlobalScope.async {
throw ArithmeticException()
}
joinAll(job, deferred)
}
Copy the codeIn resouces directory to create services directory, then in the services directory to create file kotlinx. Coroutines. CoroutineExceptionHandler
Fill in the directory with GlobalCoroutiesException coroutines18 package name and type. The exception. GlobalCoroutiesException
I’m ready to run main
But there are still problems
Cancellation and exception
Cancelling a coroutine throws a CancellationException, but we cannot find it and need to catch it
And if the child coroutine’s job calls cancel it doesn’t cancel the parent coroutine
fun main(a): Unit = runBlocking {
val child = launch {
try {
delay(Long.MAX_VALUE)
} catch (e: Exception) {
log("I caught an exception$e")}finally {
log("Subcoroutine canceled.")
}
}
yield()
log("Ready to cancel subcoroutine:")
child.cancelAndJoin()
yield()
log("End of parent coroutine")}Copy the codeException aggregation: what happens when more than one child coroutine throws an exception?
private val handler = CoroutineExceptionHandler { _, throwable ->
log("log: $throwable ${throwable.suppressed.get(0)}")}@DelicateCoroutinesApi
fun main(a): Unit = runBlocking {
val job = GlobalScope.launch(handler) {
launch {
try {
delay(Long.MAX_VALUE)
}
finally {
throw ArithmeticException()
}
}
launch {
try {
delay(1000)}finally {
throw ConcurrentModificationException()
}
}
}
job.join()
}
Copy the codeAs the code shows, Kotlin will place multiple exceptions in the suppressed property of the exception