مفهوم Flow در کاتلین

سلام بچه ها امیدوارم که حالتون خوب باشه .خب با هم بریم در مورد بحث شیرین و در عین حال خیلی مهم flow ها در کاتلین بحث کنیم.

خب اول از همه اینکه ببینیم اصلا flow ها چی هستند؟

خب Flow در کاتلین کروتین یک راه برای پردازش داده‌های جریانی (streaming data) است که امکان تولید، تبدیل و جمع‌آوری داده‌ها را فراهم می‌کند. برای درک بهتر، می‌توانیم Flow را با یک لوله فلزی که داده‌ها از آن عبور می‌کنند تصور کنیم. حالا بیایید به مرور اجزای اصلی Flow بپردازیم:

الف - Flow Builder (سازنده Flow):Flow Builder مسئول تولید داده‌ها و ارسال آنها در جریان است. مانند یک سخنران که وظیفه‌ی انجام یک کار و گفتن اطلاعات را دارد.
می‌توانیم از توابعی مانند ()flowOf یا {}flow برای ساخت یک Flow استفاده کنیم.برای آشنایی بیشتر با Flow Builder این پست رو نگاه بنداز.

ب - Operator (عملگر):عملگرها به تبدیل داده‌های جریانی از یک فرمت به فرمت دیگر کمک می‌کنند.
مانند یک مترجم که وظیفه‌ی ترجمه داده‌ها را بین دو زبان یا فرمت مختلف دارد.
عملگرها به ما امکان می‌دهند داده‌ها را تغییر داده و به شکلی دیگر در خروجی Flow بازگردانی کنیم.
برخی از عملگرها شامل map، filter، transform و ... می‌شوند.اینا هم در ادامه کامل توضیح خواهیم داد.

ج - Collector (جمع‌آورنده):جمع‌آورنده مسئول جمع‌آوری و دریافت داده‌های جریانی از Flow است.
مانند یک شنونده که می‌شنود و اطلاعات را دریافت می‌کند.
در کروتین، معمولاً از تابع collect که خود یک ترمینال اپراتور است برای جمع‌آوری داده‌ها استفاده می‌شود.

حالا بیایید یک مثال را بررسی کنیم و فرآیند Flow را برای دانلود تصویر در اندروید توضیح دهیم:

// Assume we have a function to download an image with progress
suspend fun downloadImage(): Flow<Int> = flow {
// Simulate download progress
(1..10).forEach {
delay(500) // Simulate delay
emit(it * 10) // Emit progress percentage
}
}

fun main() {
val downloadJob = CoroutineScope(Dispatchers.IO).launch {
downloadImage()
.map { progress -> "Download progress: $progress%" }
.collect { message ->
println(message) // Print download progress message
}
}

// Wait for download job to complete
runBlocking {
downloadJob.join()
}
}

در این مثال، یک Flow برای دانلود تصویر با پیشرفت دانلود به صورت درصدی تعریف شده است. سپس در یک CoroutineScope با استفاده از launch، این Flow دانلود فراخوانی می‌شود. سپس با استفاده از عملگر map، پیشرفت دانلود به پیام‌های متنی تبدیل می‌شود و در نهایت با استفاده از جمع‌آورنده collect، این پیام‌ها چاپ می‌شوند.

خروجی این کد صورت زیر می باشد:

Download progress: 10%
Download progress: 20%
Download progress: 30%
Download progress: 40%
Download progress: 50%
Download progress: 60%
Download progress: 70%
Download progress: 80%
Download progress: 90%
Download progress: 100%

در اینجا، ما با استفاده از Flow Builder، Operator و Collector می‌توانیم فرآیندی مانند دانلود تصویر را کنترل و مدیریت کنیم و پیشرفت آن را به کاربر نمایش دهیم.

در نتیجه Flow در کاتلین کروتین به شما امکان می‌دهد با داده‌های جریانی (streaming data) کار کنید. به طور خلاصه، Flow یک راه برای پردازش داده‌ها است که می‌توانید آن را به صورت همروند، بازگشتی و بی‌بلوکینگ در برنامه‌های خود استفاده کنید.

در Flow، داده‌ها به صورت پیوسته جریان دارند، به این معنی که آنها به طور متوالی و با تأخیرهای مختلفی تولید می‌شوند و شما می‌توانید این داده‌ها را به طور همروند دریافت و پردازش کنید. Flow به شما اجازه می‌دهد تا برنامه‌هایی را بنویسید که با مقداری بیشتر از حافظه بهبودیافته و کارایی بهتری داشته باشند. از آنجا که Flow بر پایه کاتلین کروتین است، می‌توانید از امکانات همروندی کروتین مانند ایجاد تردهای جدید، مدیریت خطا، تأخیر و غیره نیز استفاده کنید. در کاتلین کروتین، Flow‌ها بصورت پیشفرض بر روی تردهای اصلی اجرا نمی‌شوند.

به جای این کار، آنها بصورت معمول در تردهایی اجرا می‌شوند که با flowOn تعیین شده‌اند. این به شما امکان می‌دهد تا ترتیب اجرای Flow‌ها را مدیریت کرده و آنها را بر روی تردهای متفاوتی که بهینه‌تر بازیابی اطلاعات را فراهم می‌کنند، اجرا کنید.

برای مثال، اگر شما یک Flow را بر روی یک ترد پشته‌ای (IO) اجرا کنید و از آن برای دانلود فایل‌ها استفاده کنید، Flow می‌تواند داده‌های جریانی را در تردهای پشته‌ای اجرا کند که مناسب برای فعالیت‌های ورودی و خروجی است. همچنین می‌توانید از flowOn استفاده کنید تا Flow را بر روی تردهای مختلفی که با مواردی مانند Dispatchers.IO یا Dispatchers.Default تعیین شده‌اند، اجرا کنید.

در اینجا یک مثال استفاده از flowOn برای اجرای یک Flow بر روی ترد (IO) نشان داده شده است:

fun main() = runBlocking {
val flow = flow {
(1..5).forEach {
println("Emitting $it in thread ${Thread.currentThread().name}")
emit(it)
}
}

flowOn(Dispatchers.IO)
.collect {
println("Collected $it in thread ${Thread.currentThread().name}")
}
}

این کد یک Flow را ایجاد می‌کند که اعداد 1 تا 5 را تولید می‌کند و سپس از flowOn استفاده می‌کند تا اجرای Flow را بر روی ترد (IO) فراهم کند. وقتی این کد اجرا می‌شود، شما ممکن است یک خروجی مشابه با این را ببینید:

Emitting 1 in thread main @coroutine#1
Collected 1 in thread DefaultDispatcher-worker-1 @coroutine#2
Emitting 2 in thread main @coroutine#1
Collected 2 in thread DefaultDispatcher-worker-1 @coroutine#3
Emitting 3 in thread main @coroutine#1
Collected 3 in thread DefaultDispatcher-worker-1 @coroutine#4
Emitting 4 in thread main @coroutine#1
Collected 4 in thread DefaultDispatcher-worker-1 @coroutine#5
Emitting 5 in thread main @coroutine#1
Collected 5 in thread DefaultDispatcher-worker-1 @coroutine#6

همانطور که می‌بینید، این Flow بر روی ترد (IO) اجرا می‌شود که با نام DefaultDispatcher-worker-1 شناخته می‌شود.

حالا بریم مثال های بیشتری بزنیم تا مفهوم flow و کار کردن با اونها بیشتر برامون جا بیوفته

وقتی یک فلو روی ترد IO (Input/Output) اجرا می‌شود و سپس در داخل یک کوروتین که با Dispatcher.Main اجرا می‌شود، نتیجه این است که عملیاتی که باید در ترد IO انجام می‌شد، از ترد IO به ترد اصلی (Main Thread) منتقل می‌شود. این اتفاق به دلیل استفاده از flowOn در ابتدای فرآیند وقوع می‌پذیرد، که باعث می‌شود فلو روی یک ترد دیگری اجرا شود.

حالا بیایید یک مثال را بررسی کنیم:

fun main() {
runBlocking {
val job = CoroutineScope(Dispatchers.Main).launch {
// Launch a coroutine on the Main thread
val flowResult = withContext(Dispatchers.IO) {
// Switch to the IO thread to run the flow
createFlow()
.map { it * it }
.toList() // Collect the flow to a list
}
println("Result on Main thread: $flowResult")
}
job.join() // Wait for the coroutine to finish
}
}

fun createFlow(): Flow<Int> = flow {
// Emit some numbers
emit(1)
emit(2)
emit(3)
}

در این مثال، یک کوروتین در ترد اصلی (Main Thread) ایجاد می‌شود. درون این کوروتین، با استفاده از withContext(Dispatchers.IO)، عملیات ایجاد و اجرای فلو بر روی ترد IO صورت می‌گیرد. سپس نتایج فلو (مربوط به محاسبات مربوط به مترهای فلو) به ترد اصلی بازگردانده می‌شود و در نهایت چاپ می‌شود.

خروجی این کد به شکل زیر میباشد:

Result on Main thread: [1, 4, 9]

الان چند مثال بیشتری برایتان خواهم زد که از حالت‌های مختلفی از flowOn و استفاده از Flow در کوروتین استفاده می‌کنند.

مثال ۲: استفاده از flowOn برای اجرای یک فلو بر روی ترد پیش‌فرض (Default)

fun main() = runBlocking {
val job = CoroutineScope(Dispatchers.Main).launch {
val result = createFlow()
.flowOn(Dispatchers.Default)
.map { it * it }
.toList()
println("Result on Main thread: $result")
}
job.join()
}

fun createFlow(): Flow<Int> = flow {
emit(1)
emit(2)
emit(3)
}

حتی میتوانیم از flowOn برای اجرای یک فلو بر روی یک ترد خاص خود استفاده کنیم .به صورت زیر:

fun main() = runBlocking {
val customDispatcher = newSingleThreadContext("CustomThread")
val job = CoroutineScope(Dispatchers.Main).launch {
val result = createFlow()
.flowOn(customDispatcher)
.map { it * it }
.toList()
println("Result on Main thread: $result")
}
job.join()
}

fun createFlow(): Flow<Int> = flow {
emit(1)
emit(2)
emit(3)
}

خروجی کد بالا به صورت زیر میباشد :

Result on Main thread: [1, 4, 9]

در این مثال‌ها، از flowOn برای تعیین تردی که یک فلو بر روی آن اجرا خواهد شد استفاده شده است. با تغییر Dispatcher مربوطه در flowOn، می‌توانید فلو را بر روی تردهای مختلفی اجرا کنید.

اگر در یک فلو flowOn مشخص نشود، ترد پیشفرض برای اجرای فلو تردی است که برنامه اصلی در آن اجرا می‌شود. این به طور پیش‌فرض معمولاً ترد اصلی یا Main Thread است. در برنامه‌های کاتلین کروتین، تردهای پیش‌فرض برای اجرای کدها که بدون هیچگونه مشخصات خاصی فراخوانی می‌شوند، تردهایی هستند که از Dispatchers.Main استفاده می‌کنند.

به عبارت دیگر، اگر شما flowOn را بر روی یک فلو ناشناخته اجرا کنید، فلو بر روی ترد اصلی (Main Thread) اجرا می‌شود. این رفتار به طور پیش‌فرض در کروتین و در فلوهای بدون flowOn صورت می‌گیرد.

به عنوان مثال:

fun main() = runBlocking {
val job = CoroutineScope(Dispatchers.Main).launch {
val result = createFlow()
.map { it * it }
.toList()
println("Result on Main thread: $result")
}
job.join()
}

fun createFlow(): Flow<Int> = flow {
emit(1)
emit(2)
emit(3)
}

در این مثال، اگرچه flowOn مشخص نشده است، اما فلو بر روی ترد اصلی (Main Thread) اجرا می‌شود.

در این مثال، هیچ تردی برای فلو و کوروتین مشخص نشده است. فلو بصورت پیش‌فرض بر روی ترد اصلی اجرا می‌شود و کوروتین هم بصورت پیش‌فرض بر روی ترد پشته‌ای (Default Dispatcher) اجرا می‌شود:

fun main() = runBlocking {
val job = launch {
val result = createFlow()
.map { it * it }
.toList()
println("Result on Default thread: $result")
}
job.join()
}

fun createFlow(): Flow<Int> = flow {
emit(1)
emit(2)
emit(3)
}

خروجی:

Result on Default thread: [1, 4, 9]

در این مثال، همچنین فلو بصورت پیش‌فرض بر روی ترد اصلی (Main Thread) اجرا می‌شود، اما اجرای کوروتین بصورت پیش‌فرض بر روی ترد پشته‌ای (Default Dispatcher) انجام می‌شود.این دیسپچر یک thread pull با اندازه‌ای برابر با تعداد هسته‌های موجود در دستگاهی که کد شما در آن اجرا می‌شود دارد (اما حداقل دو ترد دارد).

آشنایی با Flow Builder ها در کاتلین

آشنایی با Terminal Operators در flow ها

آشنایی با Hot Flow و Cold Flow

آشنایی با StateFlow و SharedFlow<br/>