Java Лекция 11

Работа с Concurrency

Денис Син
Программа
модуля Основы работы с потоками и их устройство
Рассмотрим что такое потоки и процессы, когда,
как и зачем их использовать

Использование потоков в реальном коде

Разберем что предлагает библиотека Java для
работы с потоками

Продвинутые темы
Изучим Java memory model и посмотрим на
виртуальные потоки

2
План лекции
Примитивы синхронизации
Посмотрим, что предоставляет нам пакет
[Link]

Потокобезопасные коллекции и atomic

Разберемся как и когда использовать
коллекции из [Link]

ThreadPool и Executors
Изучим что такое и как работать с пулом
потоков

CompletableFuture
Как удобно можно запускать многопоточные
программы

3
План лекции
Примитивы синхронизации и atomic
Посмотрим, что предоставляет нам пакет
[Link]

Потокобезопасные коллекции
Разберемся как и когда использовать
коллекции из [Link]

ThreadPool и Executors
Изучим что такое и как работать с пулом
потоков

CompletableFuture
Как удобно можно запускать многопоточные
программы

4
[Link]
Locks

Зачем нам нужны какие-то

Lock есть же synchronized?

6
Locks Плюсы

Locks API дает возможность

расширять блок синхронизации

Synchronized не гарантирует fairness

tryLock() – позволяет ожидать

блокировку с timeout

7
Fairness
1 2 3

Поочередный захват блокировки

несколькими потоками

Lock

8
Locks

Паттерн использования

9
ReentrantLock

Один поток может брать

блокировку несколько раз

Реализация интерфейса Lock

Для гарантии fairness true в

конструкторе

10
ReadWriteLock

Несколько потоков могут брать Thread1

один ReadLock

ReadLock
Если какой-то поток взял WriteLock, Thread2

то другие блокируются

Thread3
Оптимальнее запросы на чтение

Thread4 WriteLock

11
ReadWriteLock

Несколько потоков могут брать Thread1

один ReadLock

ReadLock
Если какой-то поток взял WriteLock, Thread2

то другие блокируются

Thread3
Оптимальнее запросы на чтение

Thread4 WriteLock

12
ReadWriteLock

Несколько потоков могут брать

один ReadLock

Если какой-то поток взял WriteLock,

то другие блокируются

Оптимальнее запросы на чтение

13
Condition
Объект с условием, позволяет
блокировать поток

Поток блокируется до выполнения

условия

Нотифицирует другие потоки, чтобы

те разблокировались

14
Condition

Реализация блокирующей очереди

В очередь можно вставлять до

capactity элементов, дальше поток
заблокируется

Если очередь пуста то метод poll()

заблокирует поток

15
Дополнительные примитивы
синхронизации
CountDownLatch
Есть счетчик, как только он
опускается до 0 блокировка
снимается

countDown() – уменьшает
значение счетчика

await() – блокирует поток,

пока счетчик не равен 0

17
CountDownLatch
Есть счетчик, как только он
опускается до 0 блокировка
снимается

Блокируем один тред пока

другие треды не закончат
свои задачи

18
CyclicBarrier

Похож на CountDownLatch

Барьер можно
переиспользовать

19
CyclicBarrier

Похож на CountDownLatch

Барьер можно
переиспользовать

20
Semaphore
Есть счетчик, как только он
опускается до 0 блокировка
снимается

Блокируем один тред пока

другие треды не закончат
свои задачи

21
Semaphore
Есть счетчик, как только он
опускается до 0 блокировка
снимается

Блокируем один тред пока

другие треды не закончат
свои задачи

22
План лекции
Примитивы синхронизации и atomic
Посмотрим, что предоставляет нам пакет
[Link]

Потокобезопасные коллекции
Разберемся как и когда использовать
коллекции из [Link]

ThreadPool и Executors
Изучим что такое и как работать с пулом
потоков

CompletableFuture
Как удобно можно запускать многопоточные
программы

23
Atomics
Неблокирующие алгоритмы

Минусы Compare and swap Atomics и неблокирующие

синхронизации коллекции
Когда много потоков Можно использовать В Java есть Atomic переменные
конкурируют за блокировку неблокирующие алгоритмы, и специальные неблокирующие
тратится много ресурсов на их основанные на cas операциях коллекции
синхронизацию

25
Atomics

Основаны на cas операциях AtomicBoolean AtomicInteger AtomicLongArray AtomicLong

Могут быть использованы, как AtomicReferenceArray AtomicReference

AtomicIntegerArray
volatile переменные

26
Atomics

Основаны на cas операциях

Могут быть использованы, как

volatile переменные

27
Потокобезопасные коллекции
Synchronized
коллекции

[Link]…()

Возвращает коллекцию, операции

которой использоуют synchronized

Опять блокировки -> снижение

перформанса

29
CopyOnWriteArrayList

На каждую операцию модификации

Old array
add, remove создается копия списка

Хорошо работает, когда на одну New array

операцию записи приходится много

операций чтения

Read Write

Thread 1 Thread 2

30
ConcurrentHashMap

Замена HashMap Имеет атомарные Используется при

методы кэшировании
Не блокируется при чтении и
редко блокируется при записи • putIfAbsent(key, value) В бэкенд сервисах часто
• remove(key, value) используется для in memory
• replace(key, oldValue, newValue) кэша

31
ConcurrentSkipListMap

Замена TreeMap Имеет доп. методы Наследуется от

ConcurrentNavigableMap
Гарантируется среднее время
операций за O(logn) • ceilingEntry/Key
• floorEntry/Key

32
Producer consumer
Producer
consumer
Producer – создает и кладет
объекты в buffer
Produer Consumer
Consumer – читает и обрабатывает
объекты из buffer

34
Сложности

Producer и Consumer может быть Produer

много Consumer

У producer и consumer может быть

разная скорость Produer

Consumer
Нужно уметь синхронизировать
consumer, чтобы они не
Produer
обрабатывали сообщения 2 раза

35
BlockingQueue
Интерфейс, позволяющий
использовать очередь не беспокоясь о
синхронизации

put() – блокирует поток если

очередь полная

take() – блокирует поток если

очередь пустая

36
Пример с
condition

37
План лекции
Примитивы синхронизации и atomic
Посмотрим, что предоставляет нам пакет
[Link]

Потокобезопасные коллекции
Разберемся как и когда использовать
коллекции из [Link]

ThreadPool и Executors
Изучим что такое и как работать с пулом
потоков

CompletableFuture
Как удобно можно запускать многопоточные
программы

38
ThreadPool и Executors
Threads in real life

Просто скачиваем картинки

Что не нравится в этом коде?

40
Threads in real life

Приходится писать однотипный код

Каждый раз создается новый поток

Поток создавать дорого

А если бы картинок было бы 1К

41
ThreadPool
Что за бассейн?

42
ThreadPool
Набор потоков, который создается
сразу при старте

Потоки переиспользуются

После использования поток опять

кладется в pool и не создается по
новой

43
ExecutorService
Использует пул потоков

Использует блокирующую очередь

Задачи помещаются в очередь и

распределяются по потокам

44
ExecutorService
Future<?> submit(Runnable) – отправляет задачу на выполнение без
возвращаемого значения

Future<T> submit(Callable<T>) – отправляет задачу на выполнение с

возвращаемым значением

List<Future<T>> invokeAll(Collection<? Extends Callable<T>> –

отправляет несколько задач на выполнение

Future<T> invokeAny(Collection<? Extends Callable<T>> –

отправляет несколько задач на выполение и возвращает
результат одного которое закончится успешно

45
ExecutorService

[Link]() – блокирует поток, пока

задача не завершится

46
Future
boolean isDone()– выполнилась ли задача у ExecutorService

boolean cancel(boolean)– говорит ExecutorService прекратить выполнение

операции

boolean isCanceled() – отменена ли была задача

V get() – блокирует поток, пока ExecutorService не выполнит

задачу

V get(long, TimeUnit) – выбросит TimeoutException, если задача

не выполнится за переданное время

47
ExecutorService

48
Executors
FixedThreadPool – Пул с фиксированным кол – вом потоков
Класс с методами создания разных
реализаций ExecutorService
SingleThreadPool – Пул с одним потоком

Для разных целей используются свои

реализации CachedThreadPool – Кэширует потоки в рамках небольшого отрезка
времени

ScheduledThreadPool – Ставит задачу на время

49
План лекции
Примитивы синхронизации и atomic
Посмотрим, что предоставляет нам пакет
[Link]

Потокобезопасные коллекции
Разберемся как и когда использовать
коллекции из [Link]

ThreadPool и Excecutors
Изучим что такое и как работать с пулом
потоков

CompletableFuture
Как удобно можно запускать многопоточные
программы

50
CompletableFuture
CompletableFuture

Позволяет еще легче писать

ассинхронный код

Есть возможность писать с

использованием ExecutorService

52
CompletableFuture
boolean isDone()– проверяет записан ли результат задачи в
completableFuture

T get() – блокирует поток и ждет результата выполнения задачи

T join()– то же что и get(), но бросает CompletionException

T getNow(T valueIfAbsent)– возвращает значение либо

valueIfAbsent если его еще нет

V get(long, TimeUnit) – get() с timeout

53
CompletableFuture запуск задач
static <U> CompletableFuture<U> supplyAsync(Supplier<U> ) - Запускается задача с функцией supplier, и результат
выполнения записывается во фьючерс

static <U> CompletableFuture<U> supplyAsync(Supplier<U> , ExecutorService)– Запускается задача с функцией supplier, на

переданном ExecutorService

static CompletableFuture<Void> runAsync(Runnable)

static CompletableFuture<Void> runAsync(Runnbale, ExecutorService) - аналоги supplyAsync, но с Runnable

54
CompletableFuture обработка результата
CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) – функция, которая принимает значение из
первого future, обрабатывает его и записывает нового значение в результирующий future

CompletableFuture<Void> thenAccept(Consumer<? super T> block) – функция принимает значение из первого future и
обрабатывает его без возвращаемого значения

static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) – Возвращает новый future, который завершится когда
хотя-бы один из данных завершается

static CompletableFuture<Object> allOf(CompletableFuture<?>... cfs) – Возвращает новый future, который завершится когда
все из данных завершатся

55
CompletableFuture

Позволяет еще легче писать

ассинхронный код

Есть возможность писать с

использованием ExecutorService

56
CallbackHell

Происходит, когда много колбеков

передаются в лямбды

CompletableFuture делает такой код

легче для восприятия

57
CallbackHell

Происходит, когда много колбеков

передаются в лямбды

CompletableFuture делает такой код

легче для восприятия

58
Итоги Synchronized Locks
Примитивы
синхронизации

Неблокирующие Потокобезопасные
Atomics/CAS
алгоритмы коллекции

ThreadPool CompletableFuture

59
Доп. ресурсы

[Link] - статья про синхронизаторы

[Link] - статья про Locks

[Link] - статья про CompletableFuture

[Link] - статья про CompletableFuture

[Link] - обзор [Link]

[Link] - статься про ExecutorService

60
Всем больше Java