Многопоточность. Java презентация

java.util.concurrent: высокоуровневое API параллельного программирования

определить, выполняется ли поток;
join() - ожидать завершения потока;
run() - метод, содержащий код, который выполняется в данном потоке;
start() - запустить поток;
[static] sleep() - приостановить выполнение текущего потока на заданное время.

public void run() {
for (int i = 0; i < 10; i++) {
System.out.println("Hello");
}
}
}

new MyThread().start();

Запуск потока:

public void run() {
for (int i = 0; i < 10; i++) {
System.out.println("world");
}
}
}

new Thread(new MyRunnable()).start();

Запуск потока:

IncrementerThread(Counter counter) { this.counter = counter; }

public static void runExperiment(Counter counter) throws Exception {
Thread t1 = new IncrementerThread(counter);
Thread t2 = new IncrementerThread(counter);
long startTime = System.currentTimeMillis();
t1.start(); t2.start();
t1.join(); t2.join();
long elapsed = System.currentTimeMillis() - startTime;
System.out.println("counter=" + counter.getValue() + ", time elapsed(ms)=" + elapsed);
}

public void run() {
for (int i = 0; i < 100_000_000; i++) {
counter.increment();
}
}
}

IncrementerThread.runExperiment(new UnsafeCounter());

class Counter {
private int count;

public void increment() { count++; }

public int getValue() { return count; }
}

287 016, time elapsed(ms)= 33
counter = 105 950 712, time elapsed(ms)= 32
counter = 101 197 861, time elapsed(ms)= 32
counter = 100 029 825, time elapsed(ms)= 40

avg = 34.6 мс



Очень быстро, но неверно (ожидали 200 000 000).

@Override
public synchronized void increment() {
count++;
}

@Override
public synchronized int getValue() {
return count;
}
}

avg time = 29,6 с

Пример ReentrantLock

synchronized void put(Runnable job)
{
jobs.add(job);
this.notifyAll();
}

public synchronized Runnable getJob()
{
while (jobs.size()==0)
this.wait();

return jobs.remove(0);
}
}

monitor = new Object();
public void sendData() {
synchronized (monitor) {
try {
while (!someCondition) monitor.wait();
}
catch (InterruptedException ex) {}
System.out.println("Sending data...");
}
}
public void prepareData() {
synchronized (monitor) {
System.out.println("Data is ready");
monitor.notifyAll();
}
}
}

value;}
private Resource resourceA = new Resource();
private Resource resourceB = new Resource();

public int read() {
synchronized (resourceA) {
synchronized (resourceB) {
return resourceA.value + resourceB.value;
}
}
}

public void write (int a, int b) {
synchronized (resourceB) {
synchronized (resourceA) {
resourceA.value = a;
resourceB.value = b;
}
}
}
}

планировщиком потоков.

Специальные классы для выполнения атомарных операций находятся в пакете java.util.concurrent.atomic:

AtomicInteger
AtomicLong
AtomicDouble
AtomicReference

… и другие.

= new AtomicInteger(0);

@Override
public void increment() {
count.incrementAndGet();
}

@Override
public int getValue() {
return count.intValue();
}
}


avg time = 13,2 с

средством языка,
а обычным объектом с набором методов.
В этом случае критическую секцию
ограничивают операции lock() и unlock()
Вызов lock() блокирует, если Lock в данный
момент занят, поэтому удобно использовать метод
tryLock(), который сразу вернет управление
и результат
При использовании Lock не будут работать стандартные методы wait(), notify() и notifyAll(), ведь монитор как таковой не используется
Вместо них используются реализации интерфейса Condition, ассоциированные с Lock: необходимо вызвать Lock.newCondition() и уже у Condition вызывать методы await(), signal() и signalAll()
С одним Lock можно ассоциировать несколько Condition
class ReentrantLock

представлен
справа
Он гарантирует, что Lock будет отпущен в
любом случае, даже если при работе с
ресурсом будет выброшено исключение
Для synchronized этот подход неактуален – там средствами языка предоставляется гарантия, что мьютекс будет отпущен
Этот паттерн весьма полезен в любой ситуации, требующей обязательного освобождения ресурсов
Широко используются две основные реализации Lock:
ReentrantLock допускает вложенные критические секции
ReadWriteLock имеет разные механизмы блокировки на чтение и запись, позволяя уменьшить накладные расходы

count;
private Lock lock = new ReentrantLock();

@Override
public void increment() {
lock.lock();
count++;
lock.unlock();
}

@Override
public int getValue() {
lock.lock();
int value = count;
lock.unlock();
return value;
}
}

avg time = 22,4 с

То же самое через synchronized

count;
private Lock lock = new ReentrantLock();

@Override
public void increment() {
boolean locked = false;
while (!locked) {
locked = lock.tryLock();
}
if (locked) {
count++;
lock.unlock();
}
}

@Override
public int getValue() {
lock.lock();
int value = count;
lock.unlock();
return value;
}
}

avg time = 32,6 с

может только один поток.

class ReentrantReadWriteLock

int count;
private ReadWriteLock lock = new ReentrantReadWriteLock();

@Override
public void increment() {
lock.writeLock().lock();
count++;
lock.writeLock().unlock();
}

@Override
public int getValue() {
lock.readLock().lock();
int value = count;
lock.readLock().unlock();
return value;
}
}

avg time = 21,8 с

blockingPoolA = lock.newCondition();
Condition blockingPoolB = lock.newCondition();
Condition blockingPoolC = lock.newCondition();

{ blockingPoolA.await(); // ждём второй поток
// продолжаем работу
}
catch (InterruptedException ex) {}
finally {lock.unlock();}

Второй поток выполняет свою часть и будит первый поток:

lock.lock();
try {
// выполнение работы
blockingPoolA.signalAll(); //будим 1 поток
}
finally {lock.unlock();}

итераторы при этом не увидят изменений в коллекции
Эти коллекции следует использовать только когда 90+% операций являются операциями чтения
При частых операциях модификации большая коллекция способна убить производительность
Сортировка этих коллекций не поддерживается, т.к. она подразумевает O(n) операций вставки
Итераторы по этим коллекциям не поддерживают операций модификации

чем мониторы.
Semaphore
CountDownLatch
CyclicBarrier

потокам одновременно
N определяется параметром конструктора
При N=1 по действию аналогичен Lock
Fairness – гарантия очередности потоков

bq.put(25); // блокировка до удаления 24 из очереди
} catch (InterruptedException ex) {
}

очереди, удаляя его из очереди.
Если очередь пустая, блокируется.

poll() - возвращает первый объект очереди, удаляя его из очереди.
Если очередь пустая, возвращает null.

element() - возвращает первый элемент очереди, не удаляя его из очереди.
Если очередь пустая, то NoSuchElementException.

peek() - возвращает первый элемент очереди, не удаляя его из очереди.
Если очередь пустая, возвращает null.

class SimpleThreadExecutor implements Executor {
@Override
public void execute(Runnable command) {
command.run();
}
}

Использование:
Runnable runnable = new MyRunnableTask();
Executor executor = new SimpleThreadExecutor();
executor.execute(runnable);

ограничением количества потоков

Executors.newSingleThreadExecutor();
ровно один поток

Executors.newScheduledThreadPool();
можно настроить задержку запуска / повторный запуск


Все эти методы возвращают ExecutorService:

public interface ExecutorService extends Executor

(в виде Future).


class MyCallable implements Callable {
@Override
public String call() {
StringBuilder builder = new StringBuilder(str);
builder.reverse();
return builder.toString();
}
}

{
Integer v = f.get();
System.out.println(v);
} catch (InterruptedException | ExecutionException e) {}

boolean cancel(boolean mayInterrupt); Останавливает задачу.
boolean isCancelled(); Возвращает true, если задача была остановлена.
boolean isDone(); Возвращает true, если выполнение задачи завершено.
get(); Возвращает результат вызова метода call или кидает исключение, если оно было.