Производител и потребител

Код на:JavaC#PythonJavaScript

Накратко: Производител и потребител е шаблон, в който едни нишки слагат задачи в опашка с ограничено място, а други ги взимат; пълната опашка кара производителя да чака (backpressure).

Проблем

Черен петък

В 10:00 започва разпродажбата и сайтът приема по 50 поръчки в секунда. Складът опакова по 10. Има три лоши начина да се справиш с това.

Сайтът опакова сам. Клиентът натиска „Поръчай“ и чака, докато някой не опакова кашона. Сайтът е бърз колкото склада.

Сайтът слага поръчките в обикновен списък, а складът ги взима оттам. Два опаковчика могат да вземат една и съща поръчка, а друга да се загуби: това е race condition. Ако сложиш ключалка, опаковчик с празен списък трябва да проверява пак и пак, дали нещо е дошло.

Списъкът няма граница. Сайтът е доволен: никога не чака. Но всяка секунда списъкът расте с 40 поръчки. След час в него има 144 000, последната е чакала час, а паметта свършва.

Идеята

Опашка с ограничено място

Между двете страни се слага опашка. Производителят (producer) слага задачи в края ѝ, потребителите (consumers) ги взимат отначало. Това е шаблонът производител и потребител. Двете страни не се познават: сайтът не знае колко опаковчици има, а опаковчиците не знаят откъде идват поръчките.

Опашката е блокираща и с ограничено място:

• ако е празна, потребителят чака, без да върти процесора, докато не дойде нещо;
• ако е пълна, производителят чака, докато потребител не освободи място;
• ключалката е вътре в опашката: всяка поръчка отива точно при един потребител.

В Java това е BlockingQueue. ArrayBlockingQueue<>(3) е с 3 места. put чака, ако е пълна, а take чака, ако е празна. Тя е в java.util.concurrent, като ConcurrentLinkedQueue, в която събираме опакованите поръчки.

Виж го

Сайтът и опаковчиците

static final Order DONE = new Order(0, 0, false);

static void front(List<Order> orders,
        BlockingQueue<Order> queue, int packers)
        throws InterruptedException {
    for (Order order : orders) {
        queue.put(order); // чака, ако е пълна
    }
    for (int i = 0; i < packers; i++) {
        queue.put(DONE); // хапче за всеки опаковчик
    }
}

static void pack(BlockingQueue<Order> queue,
        Queue<Order> packed)
        throws InterruptedException {
    while (true) {
        Order order = queue.take(); // чака, ако няма
        if (order == DONE) return; // хапчето: спри
        packed.add(order);
    }
}

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

1. Сайтът (производител) слага поръчки в опашка с 3 места, двама опаковчици (потребители) ги взимат отпред. Тя е празна и те чакат в take.
2. Сайтът слага 1001 и 1002 в края на опашката.
3. Всеки опаковчик взима поръчката отпред и я опакова. Опаковането е по-бавно от приемането на поръчки.
4. Сайтът слага 1003, 1004 и 1005. И трите места са заети.
5. 1006 не се събира: put чака и сайтът спира да приема поръчки. Това е backpressure: бавната страна задържа бързата.
6. Опаковчик 1 приключва с 1001 и взима 1003. Освобождава се място и чакащият put пуска 1006.
7. И така нататък: опаковчиците взимат отпред, сайтът добавя отзад и в опашката никога няма повече от 3.
8. Поръчките свършиха. Сайтът слага по едно хапче за опаковчик след поръчките, затова всяка поръчка се опакова преди тях.
9. Всеки опаковчик довършва поръчката си, взима хапче и спира. Без хапчето щеше да чака в take вечно.
10. 8 от 8, всяка веднъж: 3 експресни за 90 €, 5 обикновени за 180 €. Кой коя опакова се сменя при всяко пускане, сумите не.

Поръчката:

record Order(int id, int euro, boolean express) {}

Сайтът и опаковчиците:

static final Order DONE = new Order(0, 0, false);

static void front(List<Order> orders,
        BlockingQueue<Order> queue, int packers)
        throws InterruptedException {
    for (Order order : orders) {
        queue.put(order); // чака, ако е пълна
    }
    for (int i = 0; i < packers; i++) {
        queue.put(DONE); // хапче за всеки опаковчик
    }
}

static void pack(BlockingQueue<Order> queue,
        Queue<Order> packed)
        throws InterruptedException {
    while (true) {
        Order order = queue.take(); // чака, ако няма
        if (order == DONE) return; // хапчето: спри
        packed.add(order);
    }
}

front и pack вървят в три нишки: един сайт и двама опаковчици. Сайтът слага осемте поръчки и свършва, а опаковчиците взимат, докато не получат знак да спрат.

List<Order> orders = List.of(
        new Order(1001, 25, false),
        new Order(1002, 40, true),
        new Order(1003, 15, false),
        new Order(1004, 60, false),
        new Order(1005, 30, true),
        new Order(1006, 45, false),
        new Order(1007, 20, true),
        new Order(1008, 35, false));
BlockingQueue<Order> queue = new ArrayBlockingQueue<>(3);
Queue<Order> packed = new ConcurrentLinkedQueue<>();
List<Thread> threads = List.of(
        new Thread(() -> run(() -> front(orders, queue, 2))),
        new Thread(() -> run(() -> pack(queue, packed))),
        new Thread(() -> run(() -> pack(queue, packed))));
for (Thread thread : threads) thread.start();
for (Thread thread : threads) thread.join();
report(orders, packed);

Отчетът проверява неща, които не зависят от това кой опаковчик коя поръчка е взел:

static void report(List<Order> orders, Queue<Order> packed) {
    long distinct = packed.stream().map(Order::id).distinct().count();
    System.out.println("Опаковани: " + packed.size() + " от " + orders.size());
    System.out.println("Всяка поръчка веднъж: " + (distinct == orders.size() ? "да" : "не"));
    for (boolean express : new boolean[] {true, false}) {
        List<Order> group = packed.stream().filter(o -> o.express() == express).toList();
        int euro = group.stream().mapToInt(Order::euro).sum();
        String kind = express ? "Експресни" : "Обикновени";
        System.out.println(kind + ": " + group.size() + " поръчки, " + euro + " €");
    }
}

Изход

Опаковани: 8 от 8
Всяка поръчка веднъж: да
Експресни: 3 поръчки, 90 €
Обикновени: 5 поръчки, 180 €

Изходът е един и същ при всяко пускане. Кой опаковчик колко поръчки е взел, се сменя, затова програмата не го печата.

Идеята

Как спират опаковчиците

Опаковчикът чака в take(), докато не дойде поръчка. Ако сайтът просто свърши, опаковчиците ще чакат вечно и програмата няма да свърши.

Затова сайтът слага накрая отровно хапче (poison pill): специална стойност, която не е поръчка. Опаковчик, който вземе хапчето, спира. Хапчетата са по едно за всеки опаковчик: всеки взима само едно и не го връща.

Хапчетата влизат в опашката след поръчките, затова всяка поръчка се опакова преди тях.

Тук хапчето е DONE: един обект Order, сравнен с ==, тоест по референция. Никоя истинска поръчка не е същият обект.

Идеята

Backpressure

Пълната опашка кара сайта да чака. Това не е грешка, а целта: бавната страна дърпа назад бързата. На английски това е backpressure. Сайтът не може да изпревари склада с повече от размера на опашката, паметта не расте, а последната поръчка чака най-много толкова, колкото трябва да се опаковат 3 пред нея.

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

offer(order) връща false, ако няма място. offer(order, 1, SECONDS) чака най-много секунда.

С него сайтът може да покаже „Голям интерес, опитайте след минута“ вместо да виси. Важното е, че решението е твое: поръчката не изчезва тихо.

Пробвай

Предвиди изхода

Опашка с 2 места и без опаковчици. Какво ще отпечата?

BlockingQueue<String> queue = new ArrayBlockingQueue<>(2);
System.out.println(queue.offer("1001"));
System.out.println(queue.offer("1002"));
System.out.println(queue.offer("1003"));
System.out.println(queue.poll());
System.out.println(queue.offer("1003"));
System.out.println(queue.size());

Отговор

true
true
true
1001
true
3

true
true
false
1002
true
2

true
true
false
1001
true
2

Провери

Пробвай

Попълни

Колко хапчета слага сайтът?

for (int i = 0; i < ; i++) {
    queue.put(DONE);
}

Провери

Пробвай

Намери грешката

Колега не иска сайтът да чака и сменя един ред. Програмата вече не виси никога, а отчетът понякога казва „Опаковани: 5 от 8“. На кой ред е грешката?

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

1static void front(List<Order> orders,2 BlockingQueue<Order> queue, int packers)3 throws InterruptedException {4 for (Order order : orders) {5 queue.offer(order);6 }7 for (int i = 0; i < packers; i++) {8 queue.put(DONE);9 }10}

Провери

Виж го

Нишки, Worker-и и съобщения

Опашката може да е и без граница: new LinkedBlockingQueue<>(). Тогава put никога не чака и проблемът от началото на урока се връща. Пълният пул от нишки (Executors.newFixedThreadPool) отвътре е точно такава опашка без граница: виж следващия урок.

Упражнение

Два входа, три опаковчика

Реши задачата в редактора си.

Поръчките идват от два входа: сайта (2001–2006) и мобилното приложение (3001–3006). Всеки вход е отделен производител, а опаковчиците са трима. Опашката е с 4 места. Хапчетата (по едно за опаковчик) идват чак след като и двата входа свършат. Отпечатай:

Изход

Опаковани: 12 от 12
Всяка поръчка веднъж: да
Общо: 325 €

Още задачи

• Отказ. Направи опашката с 1 място и сайта с offer(order, 10, MILLISECONDS). Брой отказаните поръчки и отпечатай „приети + отказани = 12“.
• По вид. Експресните поръчки трябва да се опаковат първи. PriorityBlockingQueue помага ли? Какво става с хапчетата?
• Верига. Добави втора опашка след опаковчиците: те слагат опакованите кашони в нея, а куриер ги взима. Колко хапчета трябват сега и кой ги слага?

Тест

Провери се

1. Опашката между сайта и опаковчиците е пълна. Какво прави put?

Чака, докато опаковчик не вземе поръчка и не освободи мястоХвърля поръчкатаУвеличава опашката с още едно място

2. Защо опашката изобщо има граница? Нали без нея сайтът никога не чака?

Защото опашките без граница не са безопасни между нишкиБез граница бавният склад не спира бързия сайт: опашката расте, докато не свърши паметта, а последната поръчка чака все по-дългоНяма причина: границата само забавя програмата

3. Два сайта слагат поръчки, трима опаковчици ги взимат. Колко хапчета трябват и кога се слагат?

Три, по едно за опаковчик, след като и двата сайта са свършили (след join)Едно: първият опаковчик, който го види, казва на другитеДве, по едно от всеки сайт, когато свърши

4. Как разбираш, че програмата с опашката е вярна, щом редът на опаковане е различен при всяко пускане?

Проверяваш неща, които не зависят от реда: всяка поръчка е опакована точно веднъж, броят и сумите по видСлагаш sleep, за да е редът винаги същиятПускаш я, докато два пъти подред не излезе един и същ ред

Провери

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

Свързани уроци

• Нужно отпреди: Race condition · Заключване · Атомарни операции и неизменими обекти · Deadlock
• Продължава в: Пул от нишки и задачи · Събития и опашки
• Задачи по темата: Нишки и race conditions