Все вопросы:
В этой статье:
(нажмите для перехода)
➤ Какие существуют основные операторы в RxJava
➤ Какие бывают и как работают синхронизованные коллекции
➤ Что означает volatile
➤ Что означает synchronized
➤ Какие проблемы могут быть в многопоточности в Java
➤ Что такое Lock / ReentrantLock
➤ Какие есть стратегии Backpressure
➤ Чем отличаются hot и cold Observables
➤ Чем volatile отличается от atomic
➤ Что такое отложенные (Deferred) корутины в Kotlin
➤ Что такое ленивые (LAZY) корутины в Kotlin
➤ Что такое Kotlin Channels
➤ Что такое Mutex, Monitor, Semaphore
➤ Какие существуют основные операторы в 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, 30switchMap():
Преобразует элемент в новый поток, но если поступает новый элемент, предыдущий поток прекращает эмиссию, и вместо него эмитируется новый поток. Это полезно, когда нужно всегда работать только с последним потоком.
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: 6flatMapIterable():
Преобразует каждый элемент в Iterable и эмитирует каждый элемент этого Iterable.
Observable.just(1, 2, 3)
.flatMapIterable(item -> Arrays.asList(item * 10, item * 100))
.subscribe(System.out::println); // Результат: 10, 100, 20, 200, 30, 300cast():
Преобразует элементы в потоках к определённому типу.
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, 6reduce():
Похож на scan(), но возвращает только конечный результат применения аккумуляторной функции.
Observable.just(1, 2, 3)
.reduce((accumulator, item) -> accumulator + item)
.subscribe(System.out::println); // Результат: 6window():
Разбивает поток на отдельные окна (подпотоки) с определённым количеством элементов или временем.
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:
10toList():
Преобразует поток данных в один список.
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, 4repeat():
Повторяет поток указанное количество раз.
Observable.just(1, 2, 3)
.repeat(2)
.subscribe(System.out::println); // Результат: 1, 2, 3, 1, 2, 3delay():
Задерживает эмиссию элементов на определённое время.
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, 4distinct():
Пропускает только уникальные элементы.
Observable.just(1, 2, 2, 3, 3, 3)
.distinct()
.subscribe(System.out::println); // 1, 2, 3take():
Эмитирует только указанное количество элементов.
Observable.just(1, 2, 3, 4)
.take(2)
.subscribe(System.out::println); // 1, 2skip():
Пропускает указанное количество элементов, а затем эмитирует остальные.
Observable.just(1, 2, 3, 4)
.skip(2)
.subscribe(System.out::println); // 3, 4debounce():
Эмитирует элемент только если прошло определённое время с последней эмиссии.
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, 2zip():
Комбинирует элементы из нескольких потоков, используя функцию.
Observable.zip(
Observable.just("A", "B"),
Observable.just("1", "2"),
(letter, number) -> letter + number
).subscribe(System.out::println); // A1, B2concat():
Эмитирует элементы из одного потока, затем переходит к следующему.
Observable.concat(
Observable.just("A", "B"),
Observable.just("1", "2")
).subscribe(System.out::println); // A, B, 1, 2combineLatest():
Эмитирует последний элемент из каждого потока каждый раз, когда один из потоков эмитирует новый элемент.
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, -1retry():
Повторяет выполнение потока при ошибке.
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.Mutex
val mutex = Mutex()
fun main() = runBlocking {
launch {
mutex.withLock {
// код, который нужно выполнить синхронно
}
}
}В этом примере, используется мьютекс Mutex из стандартной библиотеки Kotlin, который позволяет заблокировать доступ к общему ресурсу внутри блока 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 в Kotlin можно избежать, если взаимодействие между потоками происходит с использованием общих блокировок, например, с помощью 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 с помощью функции lazy и async из библиотеки kotlinx.coroutines.
Основное отличие отложенных корутин, созданных с помощью функции async, от LAZY корутин заключается в моменте создания объекта корутины. Отложенные корутины создаются немедленно, когда вызывается функция async, тогда как LAZY корутины создаются только в тот момент, когда к ним обращаются первый раз.
Как правило, 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 и withContext. В этом случае создается отложенная корутина, которая возвращает результат вычислений. Мы можем вызвать 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, происходит захват мьютекса.
Задачей мьютекса является защита объекта от доступа к нему других потоков, отличных от того, который завладел мьютексом.
Мьютекс защищает данные от повреждения в результате асинхронных изменений (состояние гонки), однако при неправильном использовании могут порождаться другие проблемы, например, взаимная блокировка или двойной захват.
Monitor:
высокоуровневый механизм взаимодействия и синхронизации процессов, обеспечивающий доступ к неразделяемым ресурсам.
Создает защитный механизм для реализации synchronized блоков.