Проектирование устойчивых архитектур сопрограмм

Что делает корутины устойчивыми?

import kotlinx.coroutines.*
// Resilience pillars:
// 1. Failure isolation (SupervisorJob)
// 2. Retry with backoff (retry operator / loop)
// 3. Timeout guards (withTimeout)
// 4. Clean teardown (finally + NonCancellable)
// 5. Observability (CoroutineExceptionHandler)
fun main() = runBlocking { println("Design for failure from the start") }

Шаблон области видимости уровня приложения

import kotlinx.coroutines.*
object AppCoroutineScope {
    private val handler = CoroutineExceptionHandler { _, e ->
        println("[AppScope] Uncaught: ${e.message}")
    }
    val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default + handler)
    fun cancel() = scope.cancel()
}

Повторные попытки с экспоненциальной задержкой

import kotlinx.coroutines.*
suspend fun <T> retryWithBackoff(
    times: Int = 3,
    initialDelay: Long = 100,
    block: suspend () -> T
): T {
    var delay = initialDelay
    repeat(times - 1) { attempt ->
        try { return block() }
        catch (e: Exception) {
            if (e is CancellationException) throw e
            println("Attempt ${attempt+1} failed, retrying in ${delay}ms")
            delay(delay)
            delay *= 2
        }
    }
    return block()
}
fun main() = runBlocking {
    var n = 0
    val result = retryWithBackoff {
        if (n++ < 2) throw RuntimeException("transient")
        "success"
    }
    println(result)
}

Шаблон размыкателя цепи

import kotlinx.coroutines.*
class CircuitBreaker(val maxFailures: Int, val cooldownMs: Long) {
    private var failures = 0
    private var openUntil = 0L
    suspend fun <T> call(block: suspend () -> T): T {
        if (System.currentTimeMillis() < openUntil) throw RuntimeException("Circuit open")
        return try {
            val result = block()
            failures = 0
            result
        } catch (e: Exception) {
            if (e is CancellationException) throw e
            if (++failures >= maxFailures) openUntil = System.currentTimeMillis() + cooldownMs
            throw e
        }
    }
}

Тайм-аут для каждого внешнего вызова

import kotlinx.coroutines.*
suspend fun fetchWithTimeout(url: String): String? = withTimeoutOrNull(3000) {
    // ktor: client.get(url).body()
    delay(100) // simulate
    "response from $url"
}
fun main() = runBlocking {
    val result = fetchWithTimeout("https://api.example.com")
    println(result ?: "Timed out")
}

Структурированная область видимости для каждой функции

import kotlinx.coroutines.*
class FeatureController {
    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Main)
    fun start() {
        scope.launch { /* load data */ }
        scope.launch { /* subscribe to updates */ }
    }
    fun destroy() = scope.cancel()  // cancels all feature coroutines
}

Безопасность ресурсов с конструкцией использования

import kotlinx.coroutines.*
class Connection : AutoCloseable {
    override fun close() = println("Connection closed")
    suspend fun fetch(): String { delay(100); return "data" }
}
fun main() = runBlocking {
    val job = launch {
        Connection().use { conn ->
            println(conn.fetch())
        } // close() called even on cancellation
    }
    delay(50)
    job.cancelAndJoin()
}

Корректное завершение работы

import kotlinx.coroutines.*
class WorkQueue {
    private val scope = CoroutineScope(SupervisorJob())
    private val jobs = mutableListOf<Job>()
    fun submit(block: suspend () -> Unit) {
        jobs += scope.launch { block() }
    }
    suspend fun shutdown() {
        jobs.forEach { it.join() }  // wait for all
        scope.cancel()               // then cancel scope
    }
}
fun main() = runBlocking {
    val q = WorkQueue()
    repeat(3) { i -> q.submit { delay(50); println("Task $i done") } }
    q.shutdown()
    println("Queue shut down cleanly")
}

Наблюдаемость: CoroutineName

import kotlinx.coroutines.*
fun main() = runBlocking {
    val handler = CoroutineExceptionHandler { ctx, e ->
        println("[${ctx[CoroutineName]?.name}] Failed: ${e.message}")
    }
    CoroutineScope(SupervisorJob() + handler).apply {
        launch(CoroutineName("UserLoader")) { throw RuntimeException("DB error") }
        launch(CoroutineName("Analytics")) { delay(100); println("Analytics ok") }
        delay(200); cancel()
    }
}

Избегайте GlobalScope

import kotlinx.coroutines.*
// BAD — GlobalScope leaks coroutines:
// GlobalScope.launch { delay(Long.MAX_VALUE) }

// GOOD — scoped, cancellable:
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
scope.launch { delay(100); println("Scoped") }
fun main() = runBlocking { delay(200); scope.cancel() }

Тестирование устойчивости

import kotlinx.coroutines.*
import kotlinx.coroutines.test.*
fun testRetry() = runTest {
    var attempts = 0
    val result = retryWithBackoff(3, 100) {
        if (attempts++ < 2) throw RuntimeException("fail")
        "ok"
    }
    println(result) // "ok"
}
// In test, virtual time advances instantly through delays

Быстрая проверка

Повторение