前两天在线上发现了这样的一个异常：
IllegalStateException
Flow exception transparency is violated:
Previous 'emit' call has thrown exception ${exception.e}, but then emission attempt of value '$value' has been detected.
Emissions from 'catch' blocks are prohibited in order to avoid unspecified behaviour, 'Flow.catch' operator can be used instead.
For a more detailed explanation, please refer to Flow documentation.
然后我就全局搜了一下这个错误，发现是在下面的这个方法中抛出来的：
kotlinx.coroutines.flow.internal.SafeCollector#exceptionTransparencyViolated
private fun exceptionTransparencyViolated(exception: DownstreamExceptionElement, value: Any?) {
/*
* Exception transparency ensures that if a `collect` block or any intermediate operator
* throws an exception, then no more values will be received by it.
* For example, the following code:
* ```
* val flow = flow {
* emit(1)
* try {
* emit(2)
* } catch (e: Exception) {
* emit(3)
* }
* }
* // Collector
* flow.collect { value ->
* if (value == 2) {
* throw CancellationException("No more elements required, received enough")
* } else {
* println("Collected $value")
* }
* }
* ```
* is expected to print "Collected 1" and then "No more elements required, received enough" exception,
* but if exception transparency wasn't enforced, "Collected 1" and "Collected 3" would be printed instead.
*/
error("""
Flow exception transparency is violated:
Previous 'emit' call has thrown exception ${exception.e}, but then emission attempt of value '$value' has been detected.
Emissions from 'catch' blocks are prohibited in order to avoid unspecified behaviour, 'Flow.catch' operator can be used instead.
For a more detailed explanation, please refer to Flow documentation.
""".trimIndent())
什么情况下才会有上面这样的错误呢？
我们把上面注释中的代码 copy 出来运行一下：
suspend fun main() {
val flow = flow {
emit(1)
try {
emit(2)
} catch (e: Exception) {
emit(3)
}
}
// Collector
flow.collect { value ->
if (value == 2) {
throw CancellationException("No more elements required, received enough")
} else {
println("Collected $value")
}
}
}
问题复现了。那这个问题是说什么呢？翻译一下就是：
Flow 流违反了异常透明度：
前一个 “emit” 调用已经抛出了一个 exception，但是在这之后检测到还在发射 $value。为了避免未指定的行为，禁止从 catch 块中发射数据，可以使用
Flow.catch来替代。
上面的话有些难理解，通俗来讲就是禁止在 try { emit(xx) } catch 的 catch 块中发射数据，这样的话会违反 Flow 异常的透明性。
下面我们来分析一下 Flow 的代码来看看为什么会违反 Flow 异常的透明。
flow 方法的代码如下：
public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)
注意这里的 block 函数是我们的 lamble 表达式，也就是我们调用 emit 发射数据的函数，它是 FlowCollector<T>.() -> Unit 类型的扩展函数。
flow 函数返回了一个 SafeFlow 传入了 block 函数。
接下来我们来看看 collect 的函数：
public interface Flow<out T> {
public suspend fun collect(collector: FlowCollector<T>)
}
collect 函数定义在 Flow 接口中。这里我们调用 collect 函数传入的是类型为 FlowCollector<T> 的函数式接口。上面调用 flow 返回的是 SafeFlow，所以 collect 函数的实现肯定也在 SafeFlow 这个类中。我们来看看 SafeFlow 的实现。
private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
override suspend fun collectSafely(collector: FlowCollector<T>) {
collector.block()
}
}
public abstract class AbstractFlow<T> : Flow<T>, CancellableFlow<T> {
public final override suspend fun collect(collector: FlowCollector<T>) {
val safeCollector = SafeCollector(collector, coroutineContext)
try {
collectSafely(safeCollector)
} finally {
safeCollector.releaseIntercepted()
}
}
// ...
}
SafeFlow 继承自 AbstractFlow，我们这里调用的是 AbstractFlow 的 collect 方法，这里创建了一个 SafeCollector ，然后调用了 collectorSafely 方法，在 SafeFlow 的 collectSafely 方法中又调用了 block 方法，注意这里的 block 就是调用 flow 函数传入的 lambda 代表的扩展方法。（没想到吧，还能这么玩）
所以我们调用的 emit 发射方法的实现是在 SafeCollector 中。如下：
override suspend fun emit(value: T) {
return suspendCoroutineUninterceptedOrReturn sc@{ uCont ->
try {
emit(uCont, value)
} catch (e: Throwable) {
// Save the fact that exception from emit (or even check context) has been thrown
lastEmissionContext = DownstreamExceptionElement(e)
throw e
}
}
}
emit 是一个 suspend 的方法，通过调用 suspendCoroutineUninterceptedOrReturn 方法获取到当前 suspend 函数的 Continuation ，然后调用了非 suspend 的 emit 方法。注意这里的 try-catch，如果调用非 emit 的方法出现了异常，将会把异常赋值给 lastEmissionContext，然后抛出当前捕获的异常。接下来我们来看看 非 suspend 的 emit 方法。
private fun emit(uCont: Continuation<Unit>, value: T): Any? {
val currentContext = uCont.context // 当前的协程的 context
currentContext.ensureActive() // 确保当前协程处于 active 状态（也就是没有取消，或者发生错误等）
// This check is triggered once per flow on happy path.
val previousContext = lastEmissionContext // 前一次的协程 context，如果前面 emit 发生了异常，那么这里的 previouseContext 将会是 DownstreamExceptionElement
if (previousContext !== currentContext) {
checkContext(currentContext, previousContext, value) // 检查
lastEmissionContext = currentContext
}
completion = uCont
val result = emitFun(collector as FlowCollector<Any?>, value, this as Continuation<Unit>)
/*
* If the callee hasn't suspended, that means that it won't (it's forbidden) call 'resumeWith` (-> `invokeSuspend`)
* and we don't have to retain a strong reference to it to avoid memory leaks.
*/
if (result != COROUTINE_SUSPENDED) {
completion = null
}
return result
}
private fun checkContext(
currentContext: CoroutineContext,
previousContext: CoroutineContext?,
value: T
) {
if (previousContext is DownstreamExceptionContext) {
exceptionTransparencyViolated(previousContext, value)
}
checkContext(currentContext)
}
在方法的一开始获取了当前协程的 Context，然后将 lastEmissionContext赋值给 previouseContext 。如果当前的 context 的引用不等于前一个 context 的引用（注意这里用的是 !==），那么就会调用 checkContext 来进行检查，在检查完成后会把 lastEmissionContext 赋值未当前的 currentContext。
在 checkContext 中，如果 previousContext 是 DownstreamExceptionContext 那么就会调用 exceptionTransparencyViolated 来抛出我们最开始提到的异常了。
private fun exceptionTransparencyViolated(exception: DownstreamExceptionContext, value: Any?) {
/*
* Exception transparency ensures that if a `collect` block or any intermediate operator
* throws an exception, then no more values will be received by it.
* For example, the following code:
* ```
* val flow = flow {
* emit(1)
* try {
* emit(2)
* } catch (e: Exception) {
* emit(3)
* }
* }
* // Collector
* flow.collect { value ->
* if (value == 2) {
* throw CancellationException("No more elements required, received enough")
* } else {
* println("Collected $value")
* }
* }
* ```
* is expected to print "Collected 1" and then "No more elements required, received enough" exception,
* but if exception transparency wasn't enforced, "Collected 1" and "Collected 3" would be printed instead.
*/
error("""
Flow exception transparency is violated:
Previous 'emit' call has thrown exception ${exception.e}, but then emission attempt of value '$value' has been detected.
Emissions from 'catch' blocks are prohibited in order to avoid unspecified behaviour, 'Flow.catch' operator can be used instead.
For a more detailed explanation, please refer to Flow documentation.
""".trimIndent())
那么什么情况下会触发这个异常呢？我们可以看到官方的注释中给出了一个例子。从上面的分析中我们也可以得出结论，如果我们对 emit 方法进行 try-catch，并且在 catch 中调用 emit 发射数据那么就会抛出上面的异常了。
思考一下：为什么要抛出这样的异常？
我们来继续分析，等分析完后，你可能就会有答案了。
检查没问题的话，接下来调用了 emitFun。
@Suppress("UNCHECKED_CAST")
private val emitFun =
FlowCollector<Any?>::emit as Function3<FlowCollector<Any?>, Any?, Continuation<Unit>, Any?>
public fun interface FlowCollector<in T> {
/**
* Collects the value emitted by the upstream.
* This method is not thread-safe and should not be invoked concurrently.
*/
public suspend fun emit(value: T)
}
这里对 Kotlin 和 Java 理解不够深的话，理解起来还是比较困难的。我们逐步来拆解。emitFun 是一个变量（这里不能说是一个函数类型的变量，因为它的类型不是一个高级函数），它的类型是 Function3<FlowCollector<Any?>, Any?, Continuation<Unit>, Any?>。那么为什么能把 FlowCollector<Any?>::emit 转换为 Function3 呢？
首先从 Jvm 字节码的角度出发，Follector 的 emit 就是一个方法，emit 的参数列表中的第一个参数 this 就是 Follector，最后一个参数是 Kotlin 协程 suspend 函数添加的 Continuation，所以这里可以转换为 Funcation3。
需要注意的，这里的 collector 就是我们调用 collect 时传入的 lambda 表达式所对应的对象，所以这里相当于调用了它的 emit 函数。可能还是有点难以理解，我们把上面的程序中的 lambda 表达式全部替换成匿名内部类，再来理解一下。
suspend fun main() {
val flow = flow {
emit(1)
try {
emit(2)
} catch (e: Exception) {
emit(3)
}
}
// Collector
flow.collect(object : FlowCollector<Int> {
override suspend fun emit(value: Int) {
if (value == 2) {
throw CancellationException("No more elements required, received enough")
} else {
println("Collected $value")
}
}
})
}
所以我们在 flow 中传入的 lambda 表达式中调用 emit 方法是 SafeCollector 的 emit 方法，这个 emit 方法，最终会调用到我们调用 collect 传入的 FlowCollector 的 emit 方法。
那么这里就会有一个问题，我们在 flow 上游发射数据，如果对 emit 方法加 try-catch，那么可能会 catch 到下游的异常，这违反了 flow 异常的透明性。
