Saat berhadapan dengan ribuan file, pemrosesan sekuensial menjadi lambat. Coroutines Kotlin memungkinkan paralelisme ringan tanpa overhead thread OS. Pada bab ini kita akan membangun batch processor yang memproses daftar berkas secara paralel menggunakan Dispatchers.Default, membatasi concurrency dengan Semaphore, dan melaporkan progress bertahap. Pendekatan ini cocok untuk skenario seperti konversi format gambar, parsing log besar, atau import data CSV massal.

Sequential vs Parallel #

Pemrosesan sekuensial memproses satu file pada satu waktu. Pendekatan ini sederhana namun membutuhkan waktu n × t untuk n file dengan t detik per file. Pemrosesan paralel membagi beban ke beberapa coroutine, sehingga waktu total turun ke sekitar n × t / k dengan k adalah jumlah concurrency.

StrategiWaktu (8 file, 50ms)Cocok untuk
Sequential~400 msI/O sangat ringan, urutan penting
Parallel k=4~100 msI/O ringan-sedang
Parallel k=∞~50 ms + overheadBerisiko membebani sistem

Pemilihan k penting: terlalu kecil berarti paralelisme tidak optimal, terlalu besar berarti resource exhaustion. Aturan umum: k sama dengan jumlah core CPU untuk tugas CPU-bound, dan k = 2 × core untuk tugas I/O-bound.

Batasi Concurrency dengan Semaphore #

Semaphore dari kotlinx.coroutines.sync membatasi jumlah coroutine yang aktif bersamaan. Method withPermit { } menjamin hanya n coroutine yang mengeksekusi blok pada satu waktu. Pola ini lebih ringan daripada membuat fixed thread pool manual.

kotlin Kerangka parallel processing 
val semaphore = Semaphore(4)
val hasil = coroutineScope {
    daftar.map { berkas ->
        async(Dispatchers.Default) {
            semaphore.withPermit {
                prosesBerkas(berkas)
            }
        }
    }.awaitAll()
}

awaitAll() menunggu semua Deferred selesai dan mengembalikan list hasil sesuai urutan async yang dibuat. Karena itu, kita bisa memproses hasil tanpa memikirkan urutan completion.

Demo Batch Processing #

Cuplikan berikut memproses delapan berkas demo secara paralel dengan Semaphore(4). Progress dilaporkan setelah semua selesai agar output deterministik (urutan completion coroutine tidak dapat diprediksi):

kotlin 
// Demo batch processing dengan coroutines dan progress reporting.
// Memproses daftar "file" (data class) secara paralel menggunakan
// Dispatchers.Default. Progress dilaporkan setelah semua selesai
// agar output deterministik.

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit

data class Berkas(val nama: String, val ukuran: Int)

data class HasilProses(
    val index: Int,
    val nama: String,
    val sukses: Boolean,
    val langkah: Int
)

suspend fun prosesBerkas(index: Int, berkas: Berkas): HasilProses {
    // Simulasi pemrosesan: langkah sebanding dengan ukuran
    val langkah = berkas.ukuran / 100
    delay(50L) // simulasi I/O
    return HasilProses(index, berkas.nama, sukses = true, langkah = langkah)
}

fun main() = runBlocking {
    println("=== Batch Processing dengan Coroutines ===")
    println()

    val daftar = listOf(
        Berkas("laporan-01.csv", 500),
        Berkas("laporan-02.csv", 800),
        Berkas("laporan-03.csv", 300),
        Berkas("laporan-04.csv", 1200),
        Berkas("laporan-05.csv", 600),
        Berkas("laporan-06.csv", 450),
        Berkas("laporan-07.csv", 900),
        Berkas("laporan-08.csv", 700)
    )

    println("Total berkas: ${daftar.size}")
    println("Strategi: paralel dengan Dispatchers.Default (concurrency = 4)")
    println()

    // Batasi concurrency dengan Semaphore agar tidak membebani sistem
    val semaphore = Semaphore(4)
    val total = daftar.size

    val hasil: List<HasilProses> = coroutineScope {
        daftar.mapIndexed { index, berkas ->
            async(Dispatchers.Default) {
                semaphore.withPermit {
                    prosesBerkas(index, berkas)
                }
            }
        }.awaitAll()
    }

    // Tampilkan progress berurutan berdasarkan index (deterministik)
    println("=== Progress (simulasi pelaporan bertahap) ===")
    hasil.sortedBy { it.index }.forEachIndexed { urutan, h ->
        val persen = (urutan + 1) * 100 / total
        val bar = "#".repeat(persen / 10).padEnd(10, '-')
        println("[$bar] $persen% (${urutan + 1}/$total) — selesai: ${h.nama}")
    }
    println()

    println("=== Ringkasan Hasil ===")
    hasil.sortedBy { it.nama }.forEach { h ->
        println("  ${h.nama}: sukses=${h.sukses}, langkah=${h.langkah}")
    }

    val totalLangkah = hasil.sumOf { it.langkah }
    val rataRata = if (hasil.isNotEmpty()) totalLangkah / hasil.size else 0
    println()
    println("Total langkah diproses: $totalLangkah")
    println("Rata-rata langkah per berkas: $rataRata")
    println("Berkas sukses: ${hasil.count { it.sukses }}/${hasil.size}")
}

Output yang Dihasilkan:

text 
=== Batch Processing dengan Coroutines ===

Total berkas: 8
Strategi: paralel dengan Dispatchers.Default (concurrency = 4)

=== Progress (simulasi pelaporan bertahap) ===
[#---------] 12% (1/8) — selesai: laporan-01.csv
[##--------] 25% (2/8) — selesai: laporan-02.csv
[###-------] 37% (3/8) — selesai: laporan-03.csv
[#####-----] 50% (4/8) — selesai: laporan-04.csv
[######----] 62% (5/8) — selesai: laporan-05.csv
[#######---] 75% (6/8) — selesai: laporan-06.csv
[########--] 87% (7/8) — selesai: laporan-07.csv
[##########] 100% (8/8) — selesai: laporan-08.csv

=== Ringkasan Hasil ===
  laporan-01.csv: sukses=true, langkah=5
  laporan-02.csv: sukses=true, langkah=8
  laporan-03.csv: sukses=true, langkah=3
  laporan-04.csv: sukses=true, langkah=12
  laporan-05.csv: sukses=true, langkah=6
  laporan-06.csv: sukses=true, langkah=4
  laporan-07.csv: sukses=true, langkah=9
  laporan-08.csv: sukses=true, langkah=7

Total langkah diproses: 54
Rata-rata langkah per berkas: 6
Berkas sukses: 8/8

Mari kita bedah bagian-bagian kunci dari kode di atas:

  1. data class Berkas dan HasilProses: model data dipisahkan dari logika. Berkas adalah input, HasilProses adalah output yang menyimpan index agar bisa diurutkan kembali setelah parallel completion.
  2. mapIndexed: setiap berkas mendapat index stabil yang dipakai untuk mengurutkan hasil setelah awaitAll(). Tanpa index, urutan hasil mengikuti urutan completion coroutine yang non-deterministik.
  3. Semaphore(4).withPermit { }: blok prosesBerkas hanya boleh dijalankan oleh 4 coroutine secara bersamaan. Coroutine ke-5 menunggu hingga salah satu dari 4 yang aktif selesai.
  4. Dispatchers.Default: dispatcher ini memakai thread pool sejumlah core CPU. Cocok untuk tugas CPU-bound. Untuk I/O berat (network, database), gunakan Dispatchers.IO yang punya thread pool lebih besar (default 64 thread).
  5. Progress reporting di akhir: progress dicetak setelah awaitAll() selesai dan hasil diurutkan by index. Pendekatan ini menjamin output deterministik. Untuk progress real-time yang ramah terminal, kumpulkan event ke thread-safe queue dan tampilkan dengan carriage return.
  6. delay(50L): simulasi pekerjaan I/O. Dalam program nyata, ganti dengan operasi seperti Files.readAllBytes(), URL.readText(), atau HttpClient suspend.

Progress Real-Time #

Untuk CLI yang dipakai interaktif, progress bar real-time lebih informatif. Pola umum: gunakan AtomicInteger sebagai counter bersama dan \r (carriage return) untuk overwrite baris terakhir.

kotlin Progress real-time dengan AtomicInteger 
val counter = java.util.concurrent.atomic.AtomicInteger(0)
val total = daftar.size

coroutineScope {
    daftar.map { berkas ->
        async(Dispatchers.Default) {
            semaphore.withPermit {
                prosesBerkas(berkas)
                val done = counter.incrementAndGet()
                val persen = done * 100 / total
                print("\r[$persen%] $done/$total")
                System.out.flush()
            }
        }
    }.awaitAll()
}
println() // newline setelah selesai

AtomicInteger menjamin increment bersifat atomic di banyak thread. \r mengembalikan kursor ke awal baris tanpa newline, sehingga baris yang sama ditimpa dengan progress terbaru.

Error Handling dalam Batch #

Pemrosesan paralel harus menangani error per-item, bukan menggagalkan seluruh batch. Bungkus body coroutine dengan try/catch dan kembalikan HasilProses(sukses = false, ...) bila terjadi exception.

kotlin Error handling per-item 
suspend fun prosesAman(index: Int, berkas: Berkas): HasilProses {
    return try {
        val langkah = berkas.ukuran / 100
        delay(50L)
        HasilProses(index, berkas.nama, sukses = true, langkah = langkah)
    } catch (e: Exception) {
        HasilProses(index, berkas.nama, sukses = false, langkah = 0)
    }
}

Pola ini memungkinkan batch tetap berjalan meskipun satu file gagal. Pengguna dapat melaporkan file mana yang gagal dan mencoba ulang hanya file tersebut.

Referensi Lanjutan #

Selanjutnya #

Batch processing paralel sudah berfungsi. Bab berikutnya membahas scheduling: cara menjalankan tugas secara berkala dengan java.util.Timer, ScheduledExecutorService, dan Quartz untuk skenario production. Materi tersebut ada di chapter Scheduled Jobs.