Image Processing Pipeline with Exchanger

Welcome back! In previous lessons, we covered Java concurrency utilities such as CompletableFuture, Phaser, and Semaphore. Today, we’ll revisit the Exchanger class and explore how to use it in an image processing pipeline. This lesson will help you better understand how to handle synchronization between threads and how to exchange data in a producer-consumer scenario effectively.

By the end of this lesson, you’ll learn how to:

• Use Exchanger to synchronize data between threads.
• Implement a two-stage producer-consumer pattern in a multi-threaded environment.
• Understand how to handle synchronization challenges in concurrent applications.

In this lesson, we’ll simulate an image processing pipeline where one thread loads raw image data and another applies a filter. The Exchanger will ensure smooth data exchange between these two stages, allowing us to process images step by step in a synchronized way.

The Java Exchanger class provides a synchronization point where two threads can meet and swap data. This class is particularly useful when two tasks need to exchange information before continuing their execution. In many real-world applications, such as image processing pipelines, manufacturing assembly lines, or even communication systems, there are multiple stages where different workers (or threads) need to hand off data or results to the next stage.

In this lesson, we’ll simulate an image processing pipeline where one thread (ImageLoader) loads raw image data, and another thread (ImageFilter) applies a filter to that data. These two threads will synchronize using the Exchanger, ensuring they don’t move forward until both are ready to exchange the data.

The ImageLoader is responsible for simulating the loading of raw image data. After it loads the image, it passes the data to the ImageFilter for further processing. Let’s take a look at the ImageLoader implementation:

import java.util.concurrent.Exchanger;

public class ImageLoader implements Runnable {
    private final Exchanger<String> exchanger;

    public ImageLoader(Exchanger<String> exchanger) {
        this.exchanger = exchanger;
    }

    @Override
    public void run() {
        try {
            // Simulate loading raw image data
            String imageData = "RawImageData";
            System.out.println(Thread.currentThread().getName() + " loaded image.");
            
            // Pass raw data to the ImageFilter via Exchanger
            exchanger.exchange(imageData);

            // Receive the processed image data from the ImageFilter
            imageData = exchanger.exchange(null);
            System.out.println(Thread.currentThread().getName() + " received processed image: " + imageData);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println(Thread.currentThread().getName() + " was interrupted.");
        }
    }
}

In this class, the ImageLoader does the following:

• Loads raw image data: We simulate this by initializing the imageData variable with a string representing raw image data.
• Uses exchanger.exchange(): The exchange() method is where the ImageLoader passes the raw image data to the ImageFilter thread. This method blocks until the other thread (in this case, ImageFilter) is ready to receive and exchange the data.
• Receives processed image data: After ImageFilter processes the image, the ImageLoader retrieves the processed data using another exchange() call.

This setup ensures that the ImageLoader waits until ImageFilter is ready to process the data before proceeding. If either thread is not ready, the other will wait at the exchange point, ensuring synchronization.