Rendimiento, leaks y los errores que solo aparecen bajo carga
suspend no significa concurrente, GlobalScope significa sin dueño, y una sola llamada bloqueante en el event loop deshace todo el stack reactivo.
Explorar la explicación completa 8 ideas
El primer malentendido, y el más caro, es que `suspend` significa concurrente. No lo significa: una función suspendible corre secuencialmente, exactamente como una normal, y `listOf(a, b, c).map { fetch(it) }` hace tres llamadas una tras otra. La concurrencia es opcional y siempre visible — `async` + `awaitAll`, `launch` dentro de un `coroutineScope`, o `flatMapMerge` sobre un Flow. Cuando un handler que debería tardar 100 ms tarda 300, lo primero a revisar es si tres awaits independientes se escribieron como tres sentencias secuenciales.
El segundo es que una corrutina no es gratis. Es muchísimo más barata que un hilo — una corrutina suspendida es un objeto chico en el heap, así que cien mil de ellas no son nada donde cien mil hilos son imposibles — pero cada una igual carga un `Job`, un contexto y objetos de continuación. Lanzar una por elemento de una colección grande para hacer trabajo trivial es más lento que un bucle común; la ganancia viene de *esperar* barato, no de computar barato. Las corrutinas hacen escalar el trabajo ligado a I/O; no hacen nada por el trabajo ligado a CPU salvo dejarte distribuirlo explícitamente.
El tercero, y el que tira producción abajo, es bloquear el dispatcher equivocado. `Dispatchers.Default` está dimensionado según tu cantidad de CPUs, y el event loop de Netty es aproximadamente un hilo por núcleo: un solo `Thread.sleep`, una llamada JDBC, un request de `RestTemplate` o un `File.readText` en uno de esos hilos elimina una fracción significativa de toda tu capacidad mientras dure. Bajo carga no es lento, es un atasco. La salida de emergencia correcta es `withContext(Dispatchers.IO) { … }` — un pool elástico grande hecho justo para esto — y el destino correcto es un driver no bloqueante para no necesitar la salida de emergencia.
`Dispatchers.IO` no es ilimitado, y tratarlo como vertedero es su propio modo de fallo. Por defecto tiene 64 hilos (o la cantidad de CPUs si es mayor), y un subsistema que lo satura mata de hambre a todas las demás llamadas bloqueantes del proceso. `Dispatchers.IO.limitedParallelism(n)` recorta una vista acotada para un subsistema — un cliente heredado, un generador de reportes, un job de exportación — así una dependencia lenta no puede consumir todo el pool. Es un cambio de dos palabras con el radio de acción de un mamparo.
Los leaks en código de corrutinas casi siempre tienen una de dos formas. La primera es `GlobalScope`: no tiene un `Job` que nadie posea, así que nada cancela nunca a sus hijos, y una corrutina lanzada ahí sobrevive al request, a la pantalla o al objeto que la arrancó. La segunda es un scope que creaste y nunca cancelaste — un campo `CoroutineScope(...)` en un componente sin `close()`. Ambos se encuentran igual: cada `launch` debería poder rastrearse a un scope cuya vida podés nombrar en voz alta, y si no podés nombrarla, es un leak.
`runBlocking` merece su propia advertencia. Puentea código bloqueante a código suspendible aparcando el hilo llamador hasta que el cuerpo completa — lo cual es correcto en una función `main` o en un test, y un bug en cualquier lugar dentro de un request de servidor. En un hilo de event loop bloquea el loop; llamado desde dentro de otra corrutina bloquea un hilo del pool y puede deadlockear directamente si el cuerpo necesita ese mismo dispatcher acotado para avanzar. Si encontrás `runBlocking` en un handler, el arreglo no es tunearlo, es hacer `suspend` al llamador.
Vale conocer dos costos menores. **Fan-out sin límite**: `list.map { async { call(it) } }.awaitAll()` sobre una lista de diez mil elementos dispara diez mil requests simultáneos, que es una denegación de servicio que escribiste vos — acotalo con `Semaphore(k)` o `flatMapMerge(concurrency = k)`. Y **buffering sin límite**: `Channel.UNLIMITED`, `buffer(Int.MAX_VALUE)` o un replay de `MutableSharedFlow` que crece sin parar convierten un desajuste productor/consumidor de una desaceleración en un error de memoria, que es estrictamente peor porque el back-pressure te habría avisado que algo andaba mal.
Medí antes de tunear. El reloj virtual de `runTest` convierte «¿este pipeline es secuencial?» en un número exacto, así que la pregunta secuencial-versus-buffered se responde en un test y no en producción. Agregá `-Dkotlinx.coroutines.debug` para tener nombres de corrutina en los thread dumps, y nombrá tus corrutinas para que esos dumps signifiquen algo. Y cuando un handler está lento, revisá los tres sospechosos habituales en orden: awaits secuenciales que deberían ser concurrentes, una llamada bloqueante en un dispatcher chico, y un fan-out sin límite chocando contra un rate limit — en la práctica casi siempre es uno de esos tres.
Aprendamos juntos.
Pregúntame sobre este curso o por dónde empezar.
import kotlinx.coroutines.*import kotlinx.coroutines.flow.*import kotlinx.coroutines.test.*suspend fun fetch(id: Int): String { delay(100); return "r" + id }@OptIn(ExperimentalCoroutinesApi::class)fun main() = runTest {var t = currentTimeval sequential = (1..3).map { fetch(it) } // one after anotherprintln("sequential " + sequential + " in " + (currentTime - t) + "ms")t = currentTimeval concurrent = coroutineScope { // opt in, explicitly(1..3).map { id -> async { fetch(id) } }.awaitAll()}println("concurrent " + concurrent + " in " + (currentTime - t) + "ms")t = currentTimeval streamed = (1..3).asFlow().flatMapMerge(concurrency = 3) { id -> flow { emit(fetch(id)) } }.toList()println("flatMapMerge " + streamed.sorted() + " in " + (currentTime - t) + "ms")}
suspend is not concurrent. The same three calls, written three ways, with the virtual clock reporting the difference exactly.
Arena IDEimport kotlinx.coroutines.*import kotlinx.coroutines.sync.*import kotlinx.coroutines.test.*@OptIn(ExperimentalCoroutinesApi::class)fun main() = runTest {suspend fun run(limit: Int?): String {val gate = limit?.let { Semaphore(it) }var inFlight = 0; var peak = 0val start = currentTimecoroutineScope {repeat(12) {launch {suspend fun body() {inFlight++; if (inFlight > peak) peak = inFlightdelay(100)inFlight--}if (gate == null) body() else gate.withPermit { body() }}}}return "peak=" + peak + " elapsed=" + (currentTime - start)}println("unbounded -> " + run(null)) // 12 simultaneous callsprintln("Semaphore(4) -> " + run(4)) // four at a time, three waves}
Unbounded fan-out is a self-inflicted denial of service. Semaphore(k) turns it into a load you chose, and the peak counter proves the cap held.
Arena IDEimport kotlinx.coroutines.*@OptIn(DelicateCoroutinesApi::class)fun main() = runBlocking {var owned = 0var leaked = 0val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)// Owned: cancelling the scope stops it.scope.launch { repeat(10) { delay(20); owned++ } }// Leaked: GlobalScope has no Job anyone owns, so nothing cancels it.GlobalScope.launch { repeat(10) { delay(20); leaked++ } }delay(50)scope.cancel() // "shut down the component"val ownedAtShutdown = owneddelay(150)println("owned stopped at " + ownedAtShutdown + ", still " + owned + " after 150ms")println("leaked kept going to " + leaked + " — nothing could stop it")}
GlobalScope has no owner, so cancelling the scope you DO own leaves it running. Every launch should trace back to a lifetime you can name.
Arena IDE@Serviceclass ReportService(private val jdbc: JdbcTemplate,// INJECT the dispatcher: a test can substitute StandardTestDispatcher and// get virtual time. A hard-coded Dispatchers.IO cannot be tested that way.private val io: CoroutineDispatcher = Dispatchers.IO,) {// A bulkhead: this legacy client gets 8 threads, not all 64, so a slow// provider cannot starve every other blocking call in the process.private val legacy = Dispatchers.IO.limitedParallelism(8)// WRONG on WebFlux: JDBC pins a Netty event-loop thread for the whole query.suspend fun badReport(id: Long): Report =buildReport(jdbc.queryForList("SELECT * FROM big WHERE id = ?", id))// Right: the blocking call runs on a pool sized for waiting.suspend fun report(id: Long): Report = withContext(io) {buildReport(jdbc.queryForList("SELECT * FROM big WHERE id = ?", id))}suspend fun legacyLookup(id: Long): Quote = withContext(legacy) {legacyClient.blockingQuote(id)}}
The escape hatch and the bulkhead: offload blocking work to IO, and cap a noisy subsystem so it cannot consume the whole pool. Reference only (needs Spring).
🧠 Comprueba tu comprensión
0/1 · 0/1 answered1. A WebFlux handler awaits three independent downstream services and takes ~300 ms when each service takes ~100 ms. The code is fully `suspend`, with no blocking calls. What is the most likely cause?