Skip to content
wannacode

Producer and consumer

In short: Producer-consumer is a pattern in which some threads put tasks into a bounded queue and others take them out; a full queue makes the producer wait (backpressure).

Problem

Black Friday

At 10:00 the sale starts and the site takes 50 orders a second. The warehouse packs 10. There are three bad ways to deal with that.

The site packs by itself. The customer clicks "Order" and waits until somebody packs the box. The site is as fast as the warehouse.

The site puts the orders in a plain list and the warehouse takes them from there. Two packers can take the same order and another one can get lost: that is a race condition. If you add a lock, a packer with an empty list has to check again and again whether something has come.

The list has no limit. The site is happy: it never waits. But every second the list grows by 40 orders. After an hour it holds 144 000, the last one has waited an hour, and memory runs out.

The idea

A queue with limited room

A queue goes between the two sides. The producer puts tasks at its back, the consumers take them from the front. This is the producer and consumer pattern. The two sides do not know each other: the site does not know how many packers there are, and the packers do not know where the orders come from.

The queue is blocking and has limited room:

  • if it is empty, the consumer waits, without spinning the processor, until something comes;
  • if it is full, the producer waits until a consumer frees a place;
  • the lock is inside the queue: every order goes to exactly one consumer.
In Java this is a BlockingQueue. ArrayBlockingQueue<>(3) has 3 places. put waits if it is full, and take waits if it is empty. It lives in java.util.concurrent, like ConcurrentLinkedQueue, where we collect the packed orders.

See it

The site and the packers

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); // waits if full
    }
    for (int i = 0; i < packers; i++) {
        queue.put(DONE); // one pill per packer
    }
}


static void pack(BlockingQueue<Order> queue,
        Queue<Order> packed)
        throws InterruptedException {
    while (true) {
        Order order = queue.take(); // waits if empty
        if (order == DONE) return; // the pill: stop
        packed.add(order);
    }
}
Step by step, in words
  1. The site (producer) puts orders in a queue with 3 places; two packers (consumers) take them from the front. It is empty, so they wait in take.
  2. The site puts 1001 and 1002 at the back of the queue.
  3. Each packer takes the order at the front and packs it. Packing is slower than taking orders.
  4. The site puts 1003, 1004 and 1005. All 3 places are taken.
  5. 1006 does not fit: put waits, and the site stops taking orders. This is backpressure: the slow side holds back the fast one.
  6. Packer 1 is done with 1001 and takes 1003. A place frees up, and the waiting put lets 1006 in.
  7. And so on: the packers take from the front, the site adds at the back, and the queue never holds more than 3.
  8. No more orders. The site puts one pill per packer, after the orders, so every order is packed first.
  9. Each packer finishes its order, takes a pill and stops. Without the pill it would wait in take forever.
  10. 8 of 8, each once: 3 express for 90 €, 5 standard for 180 €. Who packs which changes from run to run; the totals do not.

The order:

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

The site and the packers:

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); // waits if full
    }
    for (int i = 0; i < packers; i++) {
        queue.put(DONE); // one pill per packer
    }
}


static void pack(BlockingQueue<Order> queue,
        Queue<Order> packed)
        throws InterruptedException {
    while (true) {
        Order order = queue.take(); // waits if empty
        if (order == DONE) return; // the pill: stop
        packed.add(order);
    }
}
front and pack run in three threads: one site and two packers. The site puts the eight orders and is done, and the packers take until they get a sign to stop.
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);

The report checks things that do not depend on which packer took which order:

static void report(List<Order> orders, Queue<Order> packed) {
    long distinct = packed.stream().map(Order::id).distinct().count();
    System.out.println("Packed: " + packed.size() + " of " + orders.size());
    System.out.println("Each order once: " + (distinct == orders.size() ? "yes" : "no"));
    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 ? "Express" : "Standard";
        System.out.println(kind + ": " + group.size() + " orders, " + euro + " €");
    }
}
Output
Packed: 8 of 8
Each order once: yes
Express: 3 orders, 90 €
Standard: 5 orders, 180 €

The output is the same on every run. How many orders each packer took changes, so the program does not print it.

The idea

How the packers stop

The packer waits in take() until an order comes. If the site just finishes, the packers wait forever and the program never ends.

So at the end the site puts a poison pill: a special value that is not an order. A packer that takes the pill stops. There is one pill per packer: each takes only one and does not put it back.

The pills go into the queue after the orders, so every order is packed before them.

Here the pill is DONE: one Order object, compared with ==, that is, by reference. No real order is the same object.

The idea

Backpressure

A full queue makes the site wait. That is not a bug, it is the point: the slow side pushes back on the fast one. This is called backpressure. The site cannot get ahead of the warehouse by more than the size of the queue, memory does not grow, and the last order waits at most as long as it takes to pack the 3 in front of it.

Waiting is not the only answer to a full queue. There is also a method that does not wait, but says whether it worked:

offer(order) returns false if there is no place. offer(order, 1, SECONDS) waits at most a second.

With it the site can show "High demand, try again in a minute" instead of hanging. What matters is that the decision is yours: the order does not disappear silently.

Try it

Predict the output

A queue with 2 places and no packers. What will it print?

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());
Answer

Try it

Fill in

How many pills does the site put?
for (int i = 0; i < ; i++) {
    queue.put(DONE);
}

Try it

Find the bug

A colleague does not want the site to wait and changes one line. Now the program never hangs, but the report sometimes says "Packed: 5 of 8". Which line has the bug?

Pick the line with the bug.

See it

Threads, Workers and messages

A queue can also be without a limit: new LinkedBlockingQueue<>(). Then put never waits and the problem from the start of the lesson is back. A fixed thread pool (Executors.newFixedThreadPool) is exactly such a queue without a limit inside: see the next lesson.

Exercise

Two entrances, three packers

Solve the task in your editor.

Orders come from two entrances: the site (2001–2006) and the mobile app (3001–3006). Each entrance is a separate producer, and there are three packers. The queue has 4 places. The pills (one per packer) come only after both entrances are done. Print:

Output
Packed: 12 of 12
Each order once: yes
Total: 325 €

More tasks

  • Refusal. Make the queue 1 place and the site use offer(order, 10, MILLISECONDS). Count the refused orders and print "accepted + refused = 12".
  • By kind. Express orders must be packed first. Does PriorityBlockingQueue help? What happens to the pills?
  • A chain. Add a second queue after the packers: they put the packed boxes in it, and a courier takes them. How many pills are needed now, and who puts them?

Test

Check yourself

1. The queue between the site and the packers is full. What does put do?
2. Why does the queue have a limit at all? Without one the site never waits.
3. Two sites put orders, three packers take them. How many pills are needed, and when are they put?
4. How do you know the program with the queue is right, when the packing order is different on every run?

Made by Ivo Gergov as part of Evo Tech's campaign for accessible programming education ·