Java / Kotlin Core – Многонишковост

Кои са основните оператори в RxJava

В RxJava (Reactive Extensions for Java) има множество оператори, които позволяват да се работи с потоци от данни (Observable) асинхронно и реактивно. Тези оператори помагат да се създават, трансформират, филтрират, комбинират и обработват потоци от данни.

Ето основните категории оператори и примери за използването им:

Оператори за създаване (Creating Operators):
Тези оператори се използват за създаване на потоци (Observable).

just():
Създава поток с един или няколко елемента.

Observable.just("Hello", "RxJava")
    .subscribe(System.out::println);

fromIterable():
Преобразува колекция в поток от данни.

List<String> items = Arrays.asList("One", "Two", "Three");
Observable.fromIterable(items)
    .subscribe(System.out::println);

create():
Създава поток ръчно, с помощта на логика за емитиране.

Observable.create(emitter -> {
    emitter.onNext("Data");
    emitter.onComplete();
}).subscribe(System.out::println);

range():
Емитира последователност от числа в зададен диапазон.

Observable.range(1, 5)
    .subscribe(System.out::println);

Оператори за преобразуване (Transforming Operators):
Тези оператори преобразуват данните в потока.

map():
Преобразува всеки елемент на потока.

Observable.just(1, 2, 3)
    .map(item -> item * 2)
    .subscribe(System.out::println);

flatMap():
Преобразува всеки елемент на изходния поток в нов поток и ги обединява в един общ поток (асинхронно).

Observable.just(1, 2, 3)
    .flatMap(item -> Observable.just(item * 2))
    .subscribe(System.out::println);

concatMap():
Подобен е на flatMap(), но преобразува елементите последователно (синхронно), в реда на постъпването им.

Observable.just(1, 2, 3)
    .concatMap(item -> Observable.just(item * 10))
    .subscribe(System.out::println); // 10, 20, 30

switchMap():
Преобразува елемент в нов поток, но ако постъпи нов елемент, предишният поток прекратява емитирането и вместо него се емитира новият поток. Това е полезно, когато е нужно да се работи винаги само с последния поток.

Observable.just(1, 2, 3)
    .switchMap(item -> Observable.just(item * 10))
    .subscribe(System.out::println); // Само последният елемент от новия поток

groupBy():
Разделя потока на няколко потока (групи) въз основа на някакъв критерий.

Observable.just(1, 2, 3, 4, 5, 6)
    .groupBy(item -> item % 2 == 0 ? "Even" : "Odd")
    .subscribe(groupedObservable -> {
        groupedObservable.subscribe(item ->
            System.out.println(groupedObservable.getKey() + ": " + item)
        );
    }); // Odd: 1 Even: 2 Odd: 3 Even: 4 Odd: 5 Even: 6

flatMapIterable():
Преобразува всеки елемент в Iterable и емитира всеки елемент на този Iterable.

Observable.just(1, 2, 3)
    .flatMapIterable(item -> Arrays.asList(item * 10, item * 100))
    .subscribe(System.out::println);  // Резултат: 10, 100, 20, 200, 30, 300

cast():
Преобразува елементите в потоците към определен тип.

Observable<Object> observable = Observable.just(1, "Hello", 3.14);
observable.cast(String.class)
    .subscribe(System.out::println, Throwable::printStackTrace);  // Грешка при преобразуване на типовете

buffer():
Събира елементите в буфер (например по няколко) и ги емитира като списъци.

Observable.range(1, 10)
    .buffer(3)
    .subscribe(System.out::println);  // Извежда списъците [1, 2, 3], [4, 5, 6], ...

scan():
Прилага акумулираща функция към всеки елемент и връща всеки междинен резултат.

Observable.just(1, 2, 3)
    .scan((acc, item) -> acc + item)
    .subscribe(System.out::println);  // 1, 3, 6

reduce():
Подобен е на scan(), но връща само крайния резултат от прилагането на акумулиращата функция.

Observable.just(1, 2, 3)
    .reduce((accumulator, item) -> accumulator + item)
    .subscribe(System.out::println);  // Резултат: 6

window():
Разделя потока на отделни прозорци (подпотоци) с определен брой елементи или определено време.

Observable.range(1, 10)
    .window(3)
    .subscribe(window -> {
        System.out.println("New Window:");
        window.subscribe(System.out::println);
    });

New Window:
1
2
3
New Window:
4
5
6
New Window:
7
8
9
New Window:
10

toList():
Преобразува потока от данни в един списък.

Observable.just(1, 2, 3)
    .toList()
    .subscribe(System.out::println);  // Резултат: [1, 2, 3]

toMap():
Преобразува потока от данни в Map, като ключовете се генерират с помощта на функция.

Observable.just("apple", "banana", "cherry")
    .toMap(fruit -> fruit.charAt(0))
    .subscribe(System.out::println);  // Резултат: {a=apple, b=banana, c=cherry}

startWith():
Добавя елементи в началото на потока преди емитирането на оригиналните елементи.

Observable.just(2, 3, 4)
    .startWith(1)
    .subscribe(System.out::println);  // Резултат: 1, 2, 3, 4

repeat():
Повтаря потока зададения брой пъти.

Observable.just(1, 2, 3)
    .repeat(2)
    .subscribe(System.out::println);  // Резултат: 1, 2, 3, 1, 2, 3

delay():
Забавя емитирането на елементите с определено време.

Observable.just(1, 2, 3)
    .delay(2, TimeUnit.SECONDS)
    .subscribe(System.out::println);  // Емитира елементите след 2 секунди

debounce():
Емитира елемент само ако е минало определено време от последното емитиране (популярен е за филтриране на бързо постъпващи събития, например при работа с въвеждане от потребителя).

Observable.create(emitter -> {
    emitter.onNext(1);
    Thread.sleep(300);
    emitter.onNext(2);
    Thread.sleep(100);
    emitter.onNext(3);
    Thread.sleep(400);
    emitter.onComplete();
})
.debounce(200, TimeUnit.MILLISECONDS)
.subscribe(System.out::println);  // Емитира само 1 и 3 (игнорира 2, тъй като идва твърде бързо след 1)

throttleFirst():
Емитира първия елемент от потока за зададен интервал от време и игнорира останалите елементи през този период.

Observable.interval(100, TimeUnit.MILLISECONDS)
    .throttleFirst(1, TimeUnit.SECONDS)
    .subscribe(System.out::println);

throttleLast() (или sample()):
Емитира последния елемент, който е бил генериран за зададен интервал от време.

Observable.interval(100, TimeUnit.MILLISECONDS)
    .throttleLast(1, TimeUnit.SECONDS)
    .subscribe(System.out::println);

Оператори за филтриране (Filtering Operators):
Тези оператори се използват за филтриране на данните в потока.

filter():
Пропуска само елементите, които удовлетворяват условието.

Observable.just(1, 2, 3, 4, 5)
    .filter(item -> item % 2 == 0)
    .subscribe(System.out::println);  // 2, 4

distinct():
Пропуска само уникалните елементи.

Observable.just(1, 2, 2, 3, 3, 3)
    .distinct()
    .subscribe(System.out::println);  // 1, 2, 3

take():
Емитира само зададения брой елементи.

Observable.just(1, 2, 3, 4)
    .take(2)
    .subscribe(System.out::println);  // 1, 2

skip():
Пропуска зададения брой елементи, а след това емитира останалите.

Observable.just(1, 2, 3, 4)
    .skip(2)
    .subscribe(System.out::println);  // 3, 4

debounce():
Емитира елемент само ако е минало определено време от последното емитиране.

Observable.just(1, 2, 3, 4)
    .debounce(100, TimeUnit.MILLISECONDS)
    .subscribe(System.out::println);

*Оператори за комбиниране (Combining Operators):* Тези оператори позволяват да се обединят няколко потока от данни.

merge():
Обединява няколко потока в един, като редува елементите.

Observable.merge(
    Observable.just("A", "B"),
    Observable.just("1", "2")
).subscribe(System.out::println);  // A, 1, B, 2

zip():
Комбинира елементите от няколко потока с помощта на функция.

Observable.zip(
    Observable.just("A", "B"),
    Observable.just("1", "2"),
    (letter, number) -> letter + number
).subscribe(System.out::println);  // A1, B2

concat():
Емитира елементите от един поток, след което преминава към следващия.

Observable.concat(
    Observable.just("A", "B"),
    Observable.just("1", "2")
).subscribe(System.out::println);  // A, B, 1, 2

combineLatest():
Емитира последния елемент от всеки поток всеки път, когато някой от потоците емитира нов елемент.

Observable.combineLatest(
    Observable.just("A", "B"),
    Observable.just("1", "2", "3"),
    (letter, number) -> letter + number
).subscribe(System.out::println);  // B1, B2, B3

Оператори за обработка на грешки (Error Handling Operators):
Тези оператори помагат да се управляват грешките в потока.

onErrorReturn():
Връща елемент, ако е възникнала грешка.

Observable.just(1, 2, 0)
    .map(item -> 10 / item)
    .onErrorReturn(e -> -1)
    .subscribe(System.out::println);  // 10, 5, -1

retry():
Повтаря изпълнението на потока при грешка.

Observable.just(1, 2, 0)
    .map(item -> 10 / item)
    .retry(2)
    .subscribe(System.out::println, Throwable::printStackTrace);

onErrorResumeNext():
Продължава да емитира елементи от друг поток в случай на грешка.

Observable.just(1, 2, 0)
    .map(item -> 10 / item)
    .onErrorResumeNext(Observable.just(-1))
    .subscribe(System.out::println);

Оператори за работа с времето (Time-based Operators):
Тези оператори позволяват да се управлява времето в потоците.

interval():
Емитира елементи през равни интервали от време.

Observable.interval(1, TimeUnit.SECONDS)
    .subscribe(System.out::println);

timer():
Емитира един елемент след зададено забавяне.

Observable.timer(2, TimeUnit.SECONDS)
    .subscribe(System.out::println);

Оператори за свързване (Connectable Operators):
Те позволяват студените потоци да се превърнат в горещи, тоест емитирането на данни да започне едва след извикване на специален метод.

publish():
Превръща потока в „горещ“, но емитирането на данни започва едва след извикване на connect().

ConnectableObservable<Long> connectable = Observable.interval(1, TimeUnit.SECONDS).publish();
connectable.connect();  // Стартира емитирането на данни

Какви синхронизирани колекции има и как работят

CopyOnWrite колекции:
всички операции за промяна на колекцията (add, set, remove) водят до създаване на ново копие на вътрешния масив. Така се гарантира, че при обхождане на колекцията с итератор няма да се хвърли ConcurrentModificationException.
CopyOnWriteArrayList, CopyOnWriteArraySet

Scalable Maps:
подобрени реализации на HashMap и TreeMap с по-добра поддръжка на многонишковост и мащабируемост.
ConcurrentMap, ConcurrentHashMap, ConcurrentNavigableMap, ConcurrentSkipListMap, ConcurrentSkipListSet

Non-Blocking Queues:
нишково безопасни и неблокиращи имплементации на Queue върху свързани възли (linked nodes).
ConcurrentLinkedQueue, ConcurrentLinkedDeque

Blocking Queues:
BlockingQueue, ArrayBlockingQueue, DelayQueue, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue, BlockingDeque, LinkedBlockingDeque, TransferQueue, LinkedTransferQueue

Какво означава volatile

Ключовата дума volatile в Kotlin (и в Java) се използва за деклариране на променливи, чиито стойности могат да бъдат променяни от различни нишки. Тя гарантира, че при промяна на променливата стойността ще бъде записана веднага в паметта и прочетена от паметта, а не кеширана в регистрите на процесора, което може да доведе до проблеми със синхронизацията и с видимостта на промените.
Ключовата дума volatile може да се прилага към променливи от типовете Boolean, Byte, Char, Short, Int, Long, Float, Double, както и към референтни типове.

@Volatile
private var running: Boolean = false

В този пример променливата running ще се обновява в паметта веднага след промяна от всяка нишка, а нишките, които използват стойността ѝ, ще виждат най-новата стойност в паметта.

Какво означава synchronized

ключова дума в Kotlin (и в Java), която се използва за синхронизиране на достъпа до споделени ресурси в многонишкови приложения. Когато няколко нишки се опитват да получат достъп до споделен ресурс едновременно, могат да възникнат проблеми, например неопределеност на състоянието на ресурса или повреждането му. synchronized позволява да се избегнат тези проблеми, като гарантира, че във всеки момент достъп до споделения ресурс има само една нишка.

synchronized(lock) {
    // блок от код, в който се осъществява достъп до споделения ресурс
}
@Synchronized
fun getCounter(): Int {
   return counter
}

Корутините (coroutines) не могат да бъдат synchronized, тъй като synchronized е ключова дума в Java, която се използва за синхронизиране на достъпа до споделени ресурси между няколко нишки. Вместо това за синхронизиране на достъпа до споделени ресурси между корутини в Kotlin трябва да се използват други механизми за синхронизация, като атомарни променливи (atomic variables), заключвания (locks) или мютекси (mutexes). Например може да се използва мютекс от стандартната библиотека на Kotlin:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*

val mutex = Mutex()

fun main() = runBlocking {
    launch {
        mutex.withLock {
            // код, който трябва да се изпълни синхронно
        }
    }
}

В този пример се използва мютексът Mutex от библиотеката kotlinx.coroutines, който позволява да се заключи достъпът до споделения ресурс в блока withLock. Това гарантира, че кодът в блока withLock ще се изпълнява само от една корутина във всеки момент.

Какви проблеми могат да възникнат при многонишковостта в Java

Race condition:
ситуация, при която резултатът от изпълнението на програмата зависи от това кои нишки ще се изпълнят по-бързо или по-късно.

var count = 0

fun main() {
    Thread {
        for (i in 1..100000) {
            count++
        }
    }.start()
    Thread {
        for (i in 1..100000) {
            count++
        }
    }.start()
    Thread.sleep(1000)
    println("Count: $count")
}

В този пример се създават две нишки, всяка от които увеличава променливата count с 1 един и същ брой пъти. След изпълнението на тези нишки в главната нишка се извежда стойността на променливата count. Тъй като обаче двете нишки могат да се изпълняват паралелно, резултатът зависи от това коя нишка ще приключи първа и може да е неочакван.

Например при едно стартиране на програмата резултатът може да е 199836, а при друго – 200000. Това се случва, защото променливата count се използва неатомарно (не е защитена с механизми за синхронизация) и двете нишки могат да променят стойността ѝ едновременно. В резултат стойностите на променливата могат да се презаписват и крайната стойност може да е по-малка от очакваната.

За да се избегне Race condition в такива случаи, е необходимо да се използват синхронизация и защитни механизми, като заключвания (lock) и атомарни променливи (atomic variables), които позволяват да се гарантира коректността на изпълнението на операциите и да се избегнат неочаквани резултати.

Deadlock:
възниква, когато две или повече нишки се блокират, като чакат една друга да освободят ресурсите, необходими за продължаване на изпълнението.

Неправилно използване на synchronized: synchronized може да се използва неправилно, което може да доведе до неправилен ред на изпълнение или до блокирани нишки.

За да се избегнат тези проблеми, в Kotlin могат да се използват средства за синхронизация, като mutex, lock, atomic variables и други, както и съвременни практики за многонишково програмиране, като използване на immutable data structures и ограничаване на промяната на споделените данни само в рамките на критични секции.

class Resource(private val name: String) {
    @Synchronized fun checkResource(other: Resource) {
        println("$this: checking ${other.name}")
        Thread.sleep(1000)
        other.checkResource(this)
    }
}

fun main() {
    val resource1 = Resource("Resource 1")
    val resource2 = Resource("Resource 2")
    Thread {
        resource1.checkResource(resource2)
    }.start()
    Thread {
        resource2.checkResource(resource1)
    }.start()
}

В този пример се създават два обекта от класа Resource, всеки от които е синхронизиран с помощта на анотацията @Synchronized. След това се създават две нишки, всяка от които извиква метода checkResource() за различни ресурси в различен ред. Така, ако първата нишка заключи resource1, а втората нишка заключи resource2, двете нишки ще чакат една друга и няма да могат да завършат изпълнението си, което води до Deadlock. Deadlock в този пример може да се избегне, ако взаимодействието между нишките става с помощта на общи заключвания, например чрез synchronized или lock() от стандартната библиотека на Kotlin. Може да се използват и методите wait() и notify() за управление на нишките и за избягване на блокирането.

Livelock:
ситуация, при която две или повече нишки продължават да извършват действия, за да избегнат блокиране, но не могат да завършат работата си. В резултат те се намират в безкраен цикъл, като потребяват все повече ресурси и не вършат никаква полезна работа

data class Person(val name: String, val isPolite: Boolean = true) {
    fun greet(other: Person) {
        while (true) {
            if (isPolite) {
                println("$name: After you, ${other.name}")
                Thread.sleep(1000)
                if (other.isPolite) {
                    break
                }
            } else {
                println("$name: No, please, after you, ${other.name}")
                Thread.sleep(1000)
                if (!other.isPolite) {
                    break
                }
            }
        }
    }
}

fun main() {
    val john = Person("John", true)
    val jane = Person("Jane", false)
    Thread {
        john.greet(jane)
    }.start()
    Thread {
        jane.greet(john)
    }.start()
}

В този пример се създават два обекта от класа Person, всеки от които може да бъде учтив или неучтив в зависимост от стойността на свойството isPolite. След това се създават две нишки, всяка от които извиква метода greet() за другия обект. Методът greet() използва цикъл while, за да проверява дали другият обект е учтив и в зависимост от това да продължава или да спира.

Ако и двата обекта са учтиви, те ще си предлагат един на друг да излезе първи и няма да могат да завършат разговора си. Ако и двата обекта са неучтиви, те ще отказват да излязат първи и също няма да могат да завършат разговора. Така и двете нишки ще продължават работата си в безкраен цикъл, без да вършат никаква полезна работа, което е пример за Livelock.

Livelock в този пример може да се избегне, например като се използват таймери и се ограничи времето за изпълнение на операциите, за да могат нишките да завършат работата си и да избегнат безкрайния цикъл. Може да се използват и синхронизация и заключвания, за да се осигури коректно взаимодействие между нишките.

Какво е Lock / ReentrantLock

Lock:
това е механизъм за синхронизация, който се използва за предотвратяване на конкуренцията за достъп до споделени ресурси между няколко нишки в многонишкова среда. Когато нишка използва lock, тя получава изключителен достъп до ресурса, свързан с ключалката, а другите нишки, които се опитват да получат достъп до този ресурс, се блокират, докато първата нишка не освободи lock.

ReentrantLock:
това е реализация на интерфейса Lock в Java, която позволява на нишката да придобива и освобождава lock многократно. Нарича се „реентрантен“ (повторно влизащ), защото позволява на нишката да придобие lock отново, ако тази нишка вече има достъп до заключения ресурс.
ReentrantLock позволява механизмът на заключване да се контролира по-грижливо, отколкото със синхронизираните блокове, тъй като предоставя някои допълнителни възможности, като възможност да се спират и възобновяват нишките, които чакат достъп до заключения ресурс, възможност да се зададе таймаут за чакането на достъп до ресурса и възможност да се използват няколко условни променливи за управление на достъпа до ресурса.

Какви стратегии за Backpressure има

Buffer:
буферира всички елементи, които са били изпратени, докато потребителят не е готов да ги обработи. Тази стратегия може да доведе до недостиг на памет, ако производителят генерира елементи твърде бързо или потребителят обработва елементите твърде бавно.

Drop:
отхвърля елементите, които не могат да бъдат обработени от потребителя. Тази стратегия не гарантира, че всички елементи ще бъдат обработени, но позволява да се избегне недостигът на памет.

Latest:
съхранява само последния елемент и отхвърля всички останали. Тази стратегия е подходяща за сценарии, при които значение има само последният елемент.

Error:
сигнализира за грешка, ако потокът не може да обработи данните.

Missing:
ако потокът не може да обработва данните, той просто ще ги пропуска, без предупреждение.

Каква е разликата между hot и cold Observables

Cold Observable:
Не разпраща обекти, докато на него не се е абонирал поне един абонат;
Ако observable има няколко абоната, той ще разпрати цялата последователност от обекти на всеки абонат.

Hot Observable:
Разпраща обектите, когато се появят, независимо дали има абонати;
Всеки нов абонат получава само новите обекти, а не цялата последователност.

Каква е разликата между volatile и atomic

Ключовата дума volatile и класовете Atomic* в Java се използват за осигуряване на нишкова безопасност при достъп до споделена памет.

volatile:
гарантира, че стойността на променливата винаги ще бъде прочетена от споделената памет, а не от кеша на нишката, което предотвратява грешки в синхронизацията. Освен това записът във volatile променлива също се записва директно в споделената памет, а не в кеша на нишката.

Класове Atomic:
осигуряват атомарност на операциите с променливи. Тоест гарантират, че операциите за четене и запис ще бъдат изпълнени като едно неделимо действие. Класовете Atomic са реализирани с механизми за хардуерна поддръжка, затова в някои случаи могат да бъдат по-ефективни от използването на volatile.

Като цяло, ако е нужно само да се осигури нишкова безопасност при достъп до споделена памет, може да се използва volatile. Ако обаче е необходимо да се изпълни атомарна операция върху променлива, трябва да се използват класовете Atomic*.

Какво представляват отложените (Deferred) корутини в Kotlin

Един от механизмите за работа с асинхронни операции в Kotlin с използване на корутини. Те представляват обекти, които представят резултата от изпълнението на асинхронна операция.

Отложените корутини се създават с помощта на функцията async и могат да се използват за изпълнение на дълги операции във фонова нишка, като основната нишка на приложението остава свободна.
Щом отложената корутина завърши работата си, резултатът ѝ може да бъде получен с помощта на функцията await(). За разлика от функцията join(), която блокира извикващата нишка до завършването на изпълнението на корутината, await() не блокира извикващата нишка, а връща стойността само когато тя е готова.
Ето пример за използване на отложени корутини в Kotlin:

fun loadDataAsync(): Deferred<List<Data>> = GlobalScope.async {
    // зареждане на данни от мрежата или от база данни във фонова нишка
}
fun displayData() {
    GlobalScope.launch {
        // стартираме корутина за зареждане на данните
        val deferredData = loadDataAsync()
        // ще изпълним някои действия в основната нишка
        // получаваме резултата от изпълнението на отложената корутина
        val data = deferredData.await()
        // обработваме данните
    }
}

Какво представляват ленивите (LAZY) корутини в Kotlin

LAZY корутините са начин за създаване на отложени корутини в Kotlin: корутината се създава с параметъра start = CoroutineStart.LAZY във функциите launch или async от библиотеката kotlinx.coroutines.
Основната разлика между LAZY корутините и обикновените е в момента на стартиране: обикновената корутина, създадена с launch или async, започва да се изпълнява веднага, докато LAZY корутината се създава в неактивно състояние и стартира едва при първото извикване на start(), join() или await().
Като правило LAZY корутините се използват в случаите, когато няма нужда изпълнението на корутината да започне веднага, например когато не знаем дали изпълнението ѝ изобщо ще е необходимо. Това позволява да се избегнат излишни разходи на ресурси и да се ускори работата на приложението.
Ето пример за създаване на LAZY корутина:

val lazyCoroutine: Lazy<Deferred<String>> = lazy {
    GlobalScope.async {
        // изпълнение на корутината във фонова нишка
        "Hello from coroutine!"
    }
}

fun main() {
    // изпълнение на действия в основната нишка
    // получаване на резултата от изпълнението на корутината
    val result = runBlocking { lazyCoroutine.value.await() }
    // обработка на резултата
    println(result)
}

// или

fun main() = runBlocking<Unit> {
    val lazyCoroutine = launch(start = CoroutineStart.LAZY) {
        println("Coroutine is executing")
    }
    // изпълнение на други действия в основната нишка
    lazyCoroutine.start() // стартиране на корутината
    // изпълнение на други действия в основната нишка
    lazyCoroutine.join() // изчакване на приключването на корутината
}

В този пример създаваме отложена корутина с помощта на функцията launch и ѝ подаваме параметъра start = CoroutineStart.LAZY. След това изпълняваме някакви други действия в основната нишка и едва тогава извикваме метода start() на нашата отложена корутина, за да стартираме изпълнението ѝ.
Важно е да се разбере, че ако не извикаме метода start() на нашата отложена корутина, тя никога няма да бъде изпълнена.
Можем да използваме параметъра CoroutineStart.LAZY и с функцията async. В този случай се създава отложена корутина, която връща резултат от изчисления. Можем да извикаме await() на тази корутина, за да я стартираме и да получим резултата от изпълнението.

Какво представляват Kotlin Channels

Компонент, предоставен от Kotlin Coroutines, който позволява асинхронно предаване на стойности между корутини.
Каналите (Channels) са по-високо ниво на абстракция в сравнение с примитивните синхронизиращи обекти, като заключванията и семафорите. Те позволяват да се предават стойности между корутини, като същевременно се поддържат коректен ред и безопасност в многонишкова среда.

Kotlin Channels приличат на опашки за съобщения и имат следните основни операции: изпращане (send) и получаване (receive). Съществува и възможност каналът да се затвори (close), което води до спиране на предаването на данни.
Освен това Kotlin Channels предоставят различни операции за буфериране, като ограничаване на размера на буфера и времеви таймаут за получаване на съобщение.

Като цяло Kotlin Channels позволяват асинхронното предаване на данни в приложението да се организира по-ефективно и по-безопасно.

fun main() = runBlocking {
    val channel = Channel<Int>()
    launch {
        for (x in 1..5) channel.send(x * x)
    }
    repeat(5) { println(channel.receive()) }
    println("Done!")
}
// Очакван изход: 1 4 9 16 25 Done!

В този пример създадохме Channel с помощта на Channel(), който позволява да се предават цели числа. След това стартирахме корутина, която изпраща пет съобщения в канала с метода send(). В главната нишка получаваме съобщенията от канала с метода receive() и ги извеждаме в конзолата.
Тук използвахме функцията runBlocking за стартиране на основната нишка. Това е необходимо, тъй като функцията launch стартира корутината в нова нишка

Какво представляват Mutex, Monitor, Semaphore

Semaphore:
тип заключване, което ограничава броя на нишките, които могат да влязат в зададен участък от код.

Mutex:
обект за синхронизация на нишки; в Java мютекс (монитор) има всеки обект.
Може да има 2 състояния – свободен и зает. Състоянието на мютекса не може да се управлява директно.
Ако на друга нишка е нужен достъп до променлива, защитена с мютекс, тази нишка се блокира, докато мютексът не бъде освободен.
Отличава се от семафора по това, че само нишката, която го притежава, може да го освободи.
В блок от код, отбелязан с думата synchronized, се придобива мютексът.
Задачата на мютекса е да защити обекта от достъп до него от други нишки, различни от тази, която е придобила мютекса.
Мютексът защитава данните от повреждане в резултат на асинхронни промени (състезание, race condition), но при неправилно използване могат да възникнат други проблеми, например взаимно блокиране (deadlock) или двойно придобиване.

Monitor:
механизъм от високо ниво за взаимодействие и синхронизация на процеси, който осигурява достъп до несподеляеми ресурси.
Създава защитен механизъм за реализацията на synchronized блоковете.

Copyright: Roman Kryvolapov