Нишки

Проблем

Три отчета, един по един

Програмата сваля три месечни отчета от сървъра. Тук свалянето е пауза: Thread.sleep(ms) спира програмата за толкова милисекунди. Паузата може да бъде прекъсната отвън и тогава хвърля провереното InterruptedException, затова download го хваща.

Main.java

static void download(String file, int ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        return;
    }
    System.out.println("Свалих " + file);
}

public static void main(String[] args) {
    download("март.pdf", 600);
    download("април.pdf", 200);
    download("май.pdf", 400);
    System.out.println("Готово");
}

Изход

Свалих март.pdf
Свалих април.pdf
Свалих май.pdf
Готово

Паузите са 0,6, 0,2 и 0,4 секунди, общо 1,2 секунди. През цялото време програмата само чака мрежата. Ако имаше бутон „Отказ“, той нямаше да реагира. А трите отчета не зависят един от друг и спокойно могат да се свалят едновременно.

Идеята

Процес и нишка

Когато пуснеш програмата, операционната система ѝ дава процес: своя памет, свои отворени файлове. Вътре в процеса работят една или повече нишки (threads). Нишката е поредица от команди, които се изпълняват една след друга. Досега всяка твоя програма имаше една нишка: тази, която изпълнява main.

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

• Паралелно значи наистина едновременно, на различни ядра на процесора.
• Конкурентно значи на едно ядро, като се редуват толкова бързо, че изглежда едновременно.

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

По-точно, за любопитните

• Браузърът Chrome пуска отделен процес за всеки раздел. Затова един увиснал сайт не спира останалите, но и затова браузърът ползва много памет.
• Нишки ползва всяка програма, която прави няколко неща наведнъж: играта пуска звука в една нишка, а логиката в друга; уеб сървърът обслужва всеки клиент в своя нишка.
• Цената: всяка нишка има свой стек и паметта за нея се заделя отделно. Хиляди нишки струват. Затова в истински програми се ползва басейн от нишки (последната идея в урока).

Виж го

Първата нишка

Класът Thread е нишка. На конструктора му подаваш какво да прави: обект от интерфейса Runnable, който има един метод run(). Най-кратко е с ламбда:

Thread worker = new Thread(() -> download("април.pdf", 200));
worker.start();
System.out.println("main не чака");
worker.join();
System.out.println("Отчетът е свален");

Изход

main не чака
Свалих април.pdf
Отчетът е свален

• start() пуска новата нишка и се връща веднага. main продължава, без да чака, затова „main не чака“ се отпечатва първо.
• join() кара main да изчака нишката да свърши.

download е методът от първата карта. Добре, че хваща InterruptedException сам: run() в Runnable няма throws, затова кодът в нишката не може да остави провереното изключение да излезе навън. join() също го хвърля, затова от тук нататък main е public static void main(String[] args) throws InterruptedException.

По-точно, за любопитните

Ламбдата е най-краткият начин, но не единственият. В стария код ще срещнеш и другите три:

class ReportThread extends Thread {
    @Override
    public void run() {
        System.out.println("1. наследник на Thread");
    }
}

class ReportJob implements Runnable {
    @Override
    public void run() {
        System.out.println("2. клас, който изпълнява Runnable");
    }
}

public static void main(String[] args) throws InterruptedException {
    Thread a = new ReportThread();
    Thread b = new Thread(new ReportJob());
    Thread c = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("3. анонимен клас");
        }
    });
    Thread d = new Thread(() -> System.out.println("4. ламбда"));
    for (Thread t : new Thread[] { a, b, c, d }) {
        t.start();
        t.join();
    }
}

Изход

1. наследник на Thread
2. клас, който изпълнява Runnable
3. анонимен клас
4. ламбда

Наследникът на Thread е най-старият начин. По-гъвкаво е да отделиш задачата (Runnable) от нишката: така същата задача може да се даде на басейн от нишки, а класът ти остава свободен да наследи друг клас.

Виж го

Три наведнъж

Сега трите отчета, всеки в своя нишка:

long start = System.currentTimeMillis();
Thread march = new Thread(() -> download("март.pdf", 600));
Thread april = new Thread(() -> download("април.pdf", 200));
Thread may = new Thread(() -> download("май.pdf", 400));
march.start();
april.start();
may.start();
march.join();
april.join();
may.join();
long ms = System.currentTimeMillis() - start;
System.out.println("Готово за под секунда: " + (ms < 1000));

Изход

Свалих април.pdf
Свалих май.pdf
Свалих март.pdf
Готово за под секунда: true

Нишките тръгват в реда март, април, май, но свършват по реда на паузите: април (0,2 s), май (0,4 s), март (0,6 s). Цялото чакане е колкото най-дългото, 0,6 секунди вместо 1,2.

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

Пробвай

start или run

Какво ще отпечата програмата? (вторият аргумент на Thread е името на нишката)

Thread t = new Thread(() -> System.out.println("Работи в: " + Thread.currentThread().getName()), "сваляне");
t.run();
t.start();
t.join();

Отговор

Работи в: сваляне
Работи в: main

Работи в: main
Работи в: сваляне

Работи в: сваляне
Работи в: сваляне

Провери

Идеята

Животът на нишката

Нишката минава през няколко състояния, а getState() казва в кое е:

Thread t = new Thread(() -> download("май.pdf", 500));
System.out.println(t.getState());
t.start();
Thread.sleep(100);
System.out.println(t.getState());
t.join();
System.out.println(t.getState());

Изход

NEW
TIMED_WAITING
Свалих май.pdf
TERMINATED

Състояние	Кога
NEW	създадена, още не е пусната със start()
RUNNABLE	работи или е готова да работи
BLOCKED, WAITING, TIMED_WAITING	чака: ключалка, друга нишка (join), или определено време (sleep)
TERMINATED	run() е свършил

Нишката се пуска само веднъж. Втори start() на същата нишка хвърля IllegalThreadStateException.

По-точно, за любопитните

Още методи на Thread:

• interrupt() моли нишката да спре. Ако тя спи или чака, sleep/join хвърля InterruptedException. Затова го хващаме в download.
• getName(), threadId() (по-рано getId()): име и номер на нишката. Thread.currentThread() дава нишката, в която се изпълнява кодът.
• Thread.yield() подсказва, че нишката може да отстъпи процесора на друга. Рядко има нужда от него.

Проблем

Загубените посещения

Сайтът брои посещенията. Две нишки обслужват посетители и всяка вика increment() 100 000 пъти:

Main.java

class Counter {
    private int count = 0;

    public void increment() {
        count++;
    }

    public int getCount() {
        return count;
    }
}

public static void main(String[] args) throws InterruptedException {
    Counter visits = new Counter();
    Runnable work = () -> {
        for (int i = 0; i < 100_000; i++) {
            visits.increment();
        }
    };
    Thread first = new Thread(work);
    Thread second = new Thread(work);
    first.start();
    second.start();
    first.join();
    second.join();
    System.out.println("Посещения: " + visits.getCount());
}

Очакваш 200 000. Пет пускания:

Изход (всеки път различен)

Посещения: 150329
Посещения: 138149
Посещения: 168862
Посещения: 165672
Посещения: 157704

Всеки път по-малко и всеки път различно. Причината е, че count++ не е една стъпка, а три:

int value = count;   // 1. прочети
value = value + 1;   // 2. добави 1
count = value;       // 3. запиши

class Counter {
    private int count = 5;

    public void increment() {
        count++;   // прочети, добави 1, запиши
    }

    public synchronized void safeIncrement() {
        count++;
    }

    public synchronized int getCount() {
        return count;
    }
}

Стъпка по стъпка, с думи

1. count е 5. Нишките A и B викат increment() едновременно.
2. A прочита count: 5.
3. B прочита count, преди A да е записала: също 5.
4. A добавя 1 в своето копие и записва 6.
5. B също записва 6. Две извиквания, а count е 6, не 7: едното добавяне е загубено.
6. Отначало, със synchronized: A взима ключалката, прочита 5 и записва 6.
7. B стига до метода и чака, докато A пусне ключалката.
8. B взима ключалката, прочита 6 и записва 7: нищо не е загубено.

Когато двете нишки прочетат една и съща стойност, преди някоя да е записала, едното добавяне се губи. Това се казва състезание (race condition). Дали ще стане, зависи от това кога операционната система сменя нишките, затова резултатът е различен при всяко пускане.

Идеята

synchronized: по една

Решението е тези три стъпки да ги прави само една нишка наведнъж. Думата synchronized пред метода прави точно това:

class Counter {
    private int count = 0;

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

    public synchronized int getCount() {
        return count;
    }
}

Изход

Посещения: 200000

Всеки обект има ключалка. Нишката я взима, преди да влезе в синхронизиран метод, и я пуска, когато излезе. Другата нишка, която иска същия обект, чака. Сега резултатът е 200 000 при всяко пускане. И getCount() е синхронизиран: така нишката, която чете, вижда последната записана стойност.

Ако трябва да заключиш само част от метода, има синхронизиран блок с обект за ключалка. За брояч има и готов клас: AtomicInteger (от java.util.concurrent.atomic), чийто incrementAndGet() прави трите стъпки наведнъж, без ключалка:

class Counter {
    private int count = 0;
    private final Object lock = new Object();

    public void increment() {
        synchronized (lock) {
            count++;
        }
    }

    public int getCount() {
        synchronized (lock) {
            return count;
        }
    }
}

class Counter {
    private final AtomicInteger count = new AtomicInteger();

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

    public int getCount() {
        return count.get();
    }
}

Изход

Посещения: 200000

По-точно, за любопитните

• wait() и notifyAll(): нишка в синхронизиран метод може да пусне ключалката и да заспи, докато друга не я събуди. Така кухнята чака поръчка, а касата чака кухнята да вземе предишната (задачата „производител и потребител“):

class OrderBox {
    private String order;

    public synchronized void put(String newOrder) throws InterruptedException {
        while (order != null) {
            wait();
        }
        order = newOrder;
        notifyAll();
    }

    public synchronized String take() throws InterruptedException {
        while (order == null) {
            wait();
        }
        String result = order;
        order = null;
        notifyAll();
        return result;
    }
}

public static void main(String[] args) throws InterruptedException {
    OrderBox box = new OrderBox();
    Thread kitchen = new Thread(() -> {
        try {
            for (int i = 0; i < 3; i++) {
                System.out.println("Кухнята прие " + box.take());
            }
        } catch (InterruptedException e) {
            return;
        }
    });
    kitchen.start();
    box.put("пица");
    box.put("салата");
    box.put("супа");
    kitchen.join();
}

Изход

Кухнята прие пица
Кухнята прие салата
Кухнята прие супа

wait() е в while, а не в if: нишката може да се събуди и без причина, затова проверява условието наново.

• Мъртва хватка (deadlock): две нишки чакат една друга завинаги. Офисът държи принтера и чака скенера, а рецепцията държи скенера и чака принтера (pause е sleep с хванато изключение):

Object printer = new Object();
Object scanner = new Object();

Thread office = new Thread(() -> {
    synchronized (printer) {
        pause(100);
        synchronized (scanner) { System.out.println("Офисът копира"); }
    }
});
Thread reception = new Thread(() -> {
    synchronized (scanner) {
        pause(100);
        synchronized (printer) { System.out.println("Рецепцията копира"); }
    }
});

Програмата увисва и никога не печата нищо. Начинът да го избегнеш е всички нишки да взимат ключалките в един и същ ред.

• Приоритети: setPriority от 1 до 10 подсказва колко важна е нишката. Операционната система може да не се съобрази, затова не разчитай на тях за правилността на програмата:

Thread report = new Thread(() -> download("годишен.pdf", 100));
report.setPriority(Thread.MIN_PRIORITY);   // 1; NORM_PRIORITY е 5, MAX_PRIORITY е 10
System.out.println(report.getName() + ", приоритет " + report.getPriority());

Пробвай

Два банкомата

Двама души теглят по 80 € от една сметка със 100 € в един и същ момент. Понякога и двамата получават парите, а балансът става -60 €. На кой ред е грешката?

Избери реда с грешката.

1class Account {2 private int balance = 100;34 public void withdraw(int amount) {5 if (balance >= amount) {6 balance -= amount;7 }8 }910 public int getBalance() {11 return balance;12 }13}

Провери

Идеята

Басейн от нишки и Future

Да правиш нишка за всяка задача е скъпо, а при хиляда задачи и опасно. Басейнът от нишки (thread pool) е няколко нишки, които си взимат задачи от опашка. В Java той е ExecutorService (от java.util.concurrent):

static int sizeOf(String file, int ms) {
    try {
        Thread.sleep(ms);
    } catch (InterruptedException e) {
        return -1;
    }
    return file.length() * 100;
}

String[] files = { "март.pdf", "април.pdf", "май.pdf", "юни.pdf" };
ExecutorService pool = Executors.newFixedThreadPool(2);
List<Future<Integer>> sizes = new ArrayList<>();
for (String file : files) {
    sizes.add(pool.submit(() -> sizeOf(file, 300)));
}
for (int i = 0; i < files.length; i++) {
    System.out.println(files[i] + ": " + sizes.get(i).get() + " KB");
}
pool.shutdown();

Изход

март.pdf: 800 KB
април.pdf: 900 KB
май.pdf: 700 KB
юни.pdf: 700 KB

• Executors.newFixedThreadPool(2) прави басейн от 2 нишки. Четирите задачи се редуват в тях.
• submit приема задача, която връща резултат, и веднага дава Future: обещание, че резултатът ще дойде.
• get() чака задачата да свърши и връща резултата.
• shutdown() казва на басейна да не приема нови задачи и да спре нишките, когато свършат. Без него програмата не свършва: нишките на басейна чакат още работа.

main тук има throws Exception, защото get() хвърля и ExecutionException, ако задачата е хвърлила изключение.

По-точно, за любопитните

Ако задачата не връща нищо, има execute. Тогава чакаш всички с awaitTermination след shutdown:

ExecutorService pool = Executors.newFixedThreadPool(2);
for (int i = 1; i <= 5; i++) {
    String file = "отчет-" + i + ".pdf";
    pool.execute(() -> download(file, 100));
}
pool.shutdown();
pool.awaitTermination(1, TimeUnit.MINUTES);
System.out.println("Всички свършиха.");

Изход от едно пускане

Свалих отчет-2.pdf
Свалих отчет-1.pdf
Свалих отчет-3.pdf
Свалих отчет-4.pdf
Свалих отчет-5.pdf
Всички свършиха.

Първите два отчета вървят едновременно, затова редът им се сменя от пускане на пускане. В стари примери ще видиш while (!executor.isTerminated()) { }. Това също чака, но върти процесора на празно. awaitTermination спи, докато басейнът свърши.

Пробвай

Кой резултат пръв?

„архив.zip“ се сваля 0,6 s, а „бележка.txt“ 0,1 s. Какво ще отпечата програмата?

ExecutorService pool = Executors.newFixedThreadPool(2);
Future<Integer> big = pool.submit(() -> sizeOf("архив.zip", 600));
Future<Integer> small = pool.submit(() -> sizeOf("бележка.txt", 100));
System.out.println("Голям: " + big.get());
System.out.println("Малък: " + small.get());
pool.shutdown();

Отговор

Зависи от това коя задача свърши първа

Малък: 1100
Голям: 900

Голям: 900
Малък: 1100

Провери

Упражнение

Програми с нишки

Реши задачите в редактора си. main може да има throws InterruptedException (или throws Exception за Future.get()).

1. Паралелна сума. Масивът sales има 1 000 000 продажби, sales[i] = i % 100. Раздели го на 4 равни части и сумирай всяка в своя нишка. Всяка нишка пише в свое място в масив long[] partial, затова не трябва синхронизация. Изчакай ги с join и събери частите:

Изход

Части: [12375000, 12375000, 12375000, 12375000]
Общо: 49500000

2. Сваляне с басейн. Свали шест отчета (март.pdf, април.pdf, май.pdf, юни.pdf, юли.pdf, август.pdf) с басейн от 3 нишки. Ползвай sizeOf(file, 300) от картата за басейна. Отпечатай размерите по реда на файловете, сбора и дали всичко е станало за под 1,5 секунди (едно след друго би било 1,8 s):

Изход

март.pdf: 800 KB
април.pdf: 900 KB
май.pdf: 700 KB
юни.pdf: 700 KB
юли.pdf: 700 KB
август.pdf: 1000 KB
Общо: 4800 KB
Под 1,5 секунди: true

3. Брояч без загуби. Четири нишки, всяка с по 50 000 посещения, увеличават общ брояч. Напиши брояча така, че резултатът да е винаги 200 000. Опитай първо без синхронизация и виж колко се губи, после поправи веднъж със synchronized и веднъж с AtomicInteger.

Изход

Посещения: 200000

Още задачи

• Тест за брояча. Напиши JUnit тест, който пуска 8 нишки по 10 000 увеличения и проверява 80 000 с assertEquals. Пусни го срещу несинхронизирания брояч няколко пъти. Тест, който пада понякога, е типичният знак за състезание.
• Мъртвата хватка. Пусни примера с принтера и скенера от „По-точно“ и виж как увисва. Поправи го, като и двете нишки взимат първо принтера.

Тест

Провери се

1. Каква е разликата между t.start() и t.run()?

Няма разликаstart() пуска run() в нова нишка, а run() го изпълнява веднага в текущатаrun() пуска нова нишка, а start() не

2. Защо main вика join() преди да отпечата резултата?

За да изчака нишката да свършиЗа да стартира нишкатаЗа да спре нишката веднага

3. Две нишки правят по 100 000 пъти count++ върху общо поле без синхронизация. Какво ще е count накрая?

Винаги 100 000Често по-малко от 200 000, и различно при всяко пусканеВинаги 200 000

4. Какво гарантира synchronized метод?

Че методът е по-бързЧе в даден момент само една нишка изпълнява синхронизираните методи на този обектЧе нишките се изпълняват по реда на старта

5. Пускаш три задачи в басейн и пазиш Future за всяка. Какво прави future.get()?

Пуска задачата нановоЧака задачата да свърши и връща резултата ѝВръща резултата веднага, дори да не е готов

Провери

Започни урока отначало