OOP was supposed to save software

In the 1990s, software development was reaching a breaking point. Systems were becoming too large for procedural code to manage, memory leaks were rampant, and every new feature threatened to pull down the entire architecture like a house of cards.

Object Oriented Programming (OOP) promised a silver bullet: what if we modeled software after the real world? What if code could be encapsulated into neat, reusable "objects" that safely hid their internal complexity? Combined with Java’s famous "write once, run everywhere" motto, Java quickly became the undisputed titan of enterprise software.

Java is a hybrid language: it consists of a compiler, Javac, that compiles the code to bytecode and JVM, Java Virtual Machine, whose job is to run the bytecode.

The JVM and Javac are platform-dependent, meaning they will have different executables depending on the OS (Windows, macOS, Linux) and the architecture (x86, arm, risc-v).

Java bytecode runs on any JVM.

It was developed by James Gosling at Sun Microsystems, it had a wide adoption at the beginning of the www era for its flexibility, nowadays is the foundation of a lot of backend systems, android apps and many company frameworks.

One of the main reasons of its large adoption in the enterprise world is thanks to its programming paradigm. The main feature of Java is that it's OOP, requiring programmers to embrace Object Oriented Programming.

This allows programmers to create reusable components and build projects in a modular way, favouring scalability.

In Java we define objects through classes: A class is a blueprint for creating an object, the model, while the object itself is the actual instance of that class. It features encapsulation: both data and functions, called methods, are put together in the same class.

class Example{
    var data;

    method() {
        print("Example method");
    }
}

class App{
    main() {
        Example ex = new Example();
        ex.method();
    }
}

In this article we will build a Task Execution Pipeline, a real-world Java architecture used by frameworks like Spring and Apache Camel to distribute and execute heavy workloads across multiple CPU cores.

We will first define the blueprints for our tasks, trying to abstract the concept of "task" as much as we can. Then we will implement a "category" of tasks, specifically math tasks. After that we will try to execute them in parallel.

This will be the folder tree:

pipeline/
│
├── src/
│   ├── framework/
│   │   ├── Task.java       
│   │   └── TaskResult.java       
│   │ 
│   ├── core/
│   │   ├── BaseTask.java            
│   │   ├── SquareTask.java          
│   │   └── MathPipeline.java        
│   │
│   ├── concurrency/
│   │   └── ConcurrentEngine.java 
│   │
└── └── Main.java                    

Packages

Java projects are organized in packages, which essentially reflect the folders the classes are divided into.

We define at the beginning of each file the package it belongs to:

package framework;

Access Modifiers and Scoping Rules

Before we write code inside our packages, we must establish who can see what. Java uses access modifiers to implement data hiding and control scope.

There are four visibility levels: * public: Accessible from any class in any package. * protected: Accessible within the same package, and by the class' subclasses. * Default (no keyword): Accessible only within the same package. * private: Accessible only within the defined class.

Each file can have at most 1 public class that must have the same name of the file.

While access modifiers control who can see our code, non-access modifiers control how the code behaves, how it's stored in memory, and how it can be inherited or modified.

• static: There is only one copy of that variable or method in existence, shared by all instances of that class.
• final: Stands for unchangeable.
• abstract: Used to create a template (for classes or methods) that must be extended or implemented by subclasses.

There are many other non-access modifiers in Java, such as synchronized, volatile, transient, native, which handle specific behaviors like multi-threading and serialization.

Contrarily to access modifiers we can use more of them at the same time.

When declaring a class, method or variable Java enforces a syntax order:

[Access modifier] [Non-Access modifierS] [class / Return Type / Data Type] [Name]

For multiple non-access modifiers, standard convention dictates: static -> final/abstract -> others

Some of them are not compatible but we won't list them here, one's will learn them while programming...

Classes, Objects, Encapsulation

Mathematically, a class can be viewed as an Abstract Data Type (ADT). An ADT is a theoretical model: it defines a specific domain of values and a strict set of operations allowed on those values.

We use encapsulation to prevent outside code from messing with an object's internal values, ensuring that the object remains valid and reliable.

package framework;

public class TaskResult {
    private final double data; 
    private final boolean success;

    public TaskResult(double data, boolean success) {
        this.data = data;
        this.success = success;
    }

    public double getData() { return data; }
    public boolean isSuccess() { return success; }
}

We encapsulate data with private to avoid other classes to direcly modify them, the constructor allows the programmer to check the data before assigning it.

Extending classes

Now that we have objects, how do we relate them to one another? Inheritance allows a new class (subclass) to inherit fields and methods from another existing class (superclass), promoting code reuse.

Inheritance introduces Polymorphism ("many forms"). It means a subclass object can be treated as an instance of its superclass.

Let's move to our core package and create a BaseTask class for our MathPipeline. Other specific tasks will extend this base class.

package core;
import framework.TaskResult;

public class BaseTask {
    protected double input; 

    public BaseTask(double input) {
        this.input = input;
    }

    public void logExecution() {
        System.out.println("Processing input value: " + input);
    }

    public TaskResult process() {
        return new TaskResult(0.0, true);
    }
}

Now, we extend BaseTask to create a specific SquareTask.

package core;
import framework.TaskResult;

public class SquareTask extends BaseTask {

    public SquareTask(double input) {
        super(input); 
    }

    @Override
    public TaskResult process() {
        logExecution(); 
        double result = input * input;
        return new TaskResult(result, true);
    }
}

We use the super keyword to invoke the parent constructor and "override" the process method to change its behavior.

Abstract Classes and Interfaces

Ideally, another programmer would just need to implement process() without remembering to call logExecution(), since we always want tasks logging as they are running.

We need to conceptually distinguish what a task is from what it does: "A Task is function that takes a double as input and returns a TaskResult(double, boolean)"

We want to reduce BaseTask to a minimal blueprint that allows future programmers to implement just the process method.

We define an Execute method that strips away the need for logging and allows extending the class caring only about process().

Let's refactor our TaskFramework to include a Task interface. An interface is a pure blueprint. It only defines methods, to be implemented by classes that implement that interface.

A class can implement multiple interfaces, implementing their methods.

package framework;

public interface Task {
    TaskResult execute();
}

Now, we can update our BaseTask in the core package to become abstract and implement the Task interface.

An abstract class cannot be instantiated on its own. It can hold data, contrarily to interfaces, and contains abstract methods that must be implemented by subclasses.

A class can implement multiple interfaces, since but can only inherit from one class.

package core;

import framework.*; // Import all classes from the package

public abstract class BaseTask implements Task {
    protected double input;

    public BaseTask(double input) {
        this.input = input;
    }

    public void logExecution() {
        System.out.println("Executing pipeline task with input: " + input);
    }

    public abstract TaskResult process();

    @Override
    public final TaskResult execute() {
        logExecution();
        return process(); 
    }
}

In this way we we will implement specific tasks starting from BaseTask. We "abstract" the process method, subclasses will implement it.

SquareTask will still extend BaseTask implementing process:

package core;

import framework.*;

public class SquareTask extends BaseTask {

    public SquareTask(double input) {
        super(input);
    }

    @Override
    public TaskResult process() {
        double squaredValue = this.input * this.input;

        return new TaskResult(squaredValue, true);
    }
}

We just defined what a Task is, then we defined the minimal "actions" a task should have in BaseTask (which would be logging that a task is being executed with the logExecution method) and then defined the specific actions a particular method should have in SquareTask, in the process method.

Generics and Type Hierarchies

So far we abstracted what a task is, anyone can extend our BaseClass writing a custom function implementing only the process() method and still have logging or use the TaskResult methods...

But what if a task doesn't use double as input? What if a future programmer using our code wants to return a type different than double in the TaskResult class?

We want to define a task like so: "A Task is function that takes any input and returns a TaskResult(any, boolean)" we want to leave future programmers freedom to do whatever they want with whatever input they like, they just need to return a TaskResult(any, boolean).

We could extend BaseTask as we already did but this approaches creates a rigid trap. If we only rely on extending this concrete version of BaseTask, every new task in our system is permanently forced to inherit the protected double input variable and the default process() logic.

To achieve complete data type flexibility we use Generics. Generics allow "parameterizing" data types.

package framework;

public interface Task<O> {
    TaskResult<O> execute();
}

package framework;

public class TaskResult<T> {
    private final T data;
    private final boolean success;

    public TaskResult(T data, boolean success) {
        this.data = data;
        this.success = success;
    }

    public T getData() { return data; }
    public boolean isSuccess() { return success; }
}

and are generic parameters for the actual data types.

Sometimes, we want flexibility, but with specific rules. For example, in our tasks, we should allow to process types like Integer, Double, or Float, but it shouldn't allow a String.

We can use bounded type parameters:

package core;

import framework.*;

public abstract class BaseTask<I extends Number, O> implements Task<O> {

    protected final I input;

    public BaseTask(I input) {
        this.input = input;
    }

    public void logExecution() {
        System.out.println("Executing pipeline task with input: " + input);
    }

    public abstract TaskResult<O> process();

    @Override
    public final TaskResult<O> execute() {
        logExecution();
        return process(); 
    }
}

In this way we can define the types directly in the actual implementation.

package core;

import framework.*;

public class SquareTask extends BaseTask<Double, Double> {
    public SquareTask(Double input) {
        super(input);
    }

    @Override
    public TaskResult<Double> process() {
        double squaredValue = this.input * this.input;
        return new TaskResult<>(squaredValue, true);
    }
}

Let's finally implement MathPipeline using all the structures we already have. We want to define a List of tasks to execute and add a method to create a new task, as well as a method to retrieve the oldest added Task.

package core;

import framework.*;
import java.util.ArrayList;
import java.util.List;

public class MathPipeline {
    private final List<Task<Double>> taskQueue = new ArrayList<>();

    public void addTask(Task<Double> task) {
        if (task != null) {
            taskQueue.add(task);
        }
    }

    public Task<Double> pollTask() {
        if (!taskQueue.isEmpty()) {
            return taskQueue.remove(0); // Removes and returns the first element
        }
        return null;
    }
}

MathPipeline acts as a concrete implementation of our Abstract Data Type. Mathematically, it models a First-In-First-Out (FIFO) queue. The algebraic properties of this structure ensure that the data remains consistent across operations: any task added via addTask will maintain its relative ordering when retrieved via pollTask.

Threads and Process Synchronization

Currently, our task execution pipeline runs linearly on a single thread. In the real world, computing power comes from multicore processors. To maximize efficiency, we must introduce Concurrency using threads.

A Thread is a single unit of execution within a process. Multiple threads can run simultaneously, sharing the same memory space (the Heap).

However, this shared memory introduces a critical issue: if two threads try to modify the same resource at the exact same time, the data becomes corrupted. This is known as a Race Condition.

To prevent race conditions, we use Process Synchronization to ensure that only one thread can access a critical block of code at any given moment. In Java this is done with the synchronized keyword.

In our ConcurrentEngine we will add a method to start a single worker, i.e. an object that takes tasks from our Pipeline and executes them. We will use the Thread object provided by java.

After that, in our Main, we will istantiate more workers that will work on different threads in parallel.

package concurrency;

import framework.Task;
import core.MathPipeline;
import java.util.List;

public class ConcurrentEngine {
    private final MathPipeline pipeline;

    public ConcurrentEngine(MathPipeline pipeline) {
        this.pipeline = pipeline;
    }

    public void startWorker(String workerName) {

        Thread thread = new Thread(/* Function to execute */);

        thread.setName(workerName);
        thread.start();
    }
}

We define a final MathPipeline, since it will be shared by all our threads.

In the startWorker method we create a Thread object and pass it a function. The Thread object's constructor expects a Runnable object.

(This is the original source code implementing Thread)

public class Thread implements Runnable {
    // ...
    public Thread(Runnable target) {
        this(null, target, "Thread-" + nextThreadNum(), 0);
    }
    // ...
}

We could implement the Runnable interface elsewhere, for example using an inner class inside ConcurrentEngine. Creating the object in the startWorker method and then passing it to the Thread object.

public class ConcurrentEngine {
    //...
    public void startWorker(String workerName) {
        Worker workerLogic = new Worker(workerName);
        Thread thread = new Thread(workerLogic);
        thread.setName(workerName);
        thread.start();
    }

    private class Worker implements Runnable {
        //...
        @Override
        public void run() {/* ... */}
    }
}

If another programmer reads our code, they would have to jump to look what a worker does, that's not ideal. Since our worker does minimal things, polls and executes, creating a class to call 2 methods feels overkill.

For the sake of readability it would be better to create the object inline.

Java allows creating objects inline using Anonymous classes.

Thread thread = new Thread(new Runnable() {
    @Override
    public void run() { ... }
});

But using them would require a lot of boilerplate code just to override a single run method.

In modern Java we can use Lambdas and the compiler will handle that for us.

A lambda is a nice way to define a function inline using the following syntax:

(param 1, param 2) -> {function body...};

So we can write:

Thread thread = new Thread(() -> {
    ...
});

The Thread, which is our main processing unit, will run indefinitely, polling tasks from our pipeline and executing them.

The synchronized keyword is crucial to avoid multiple workers executing the same task.

Under the hood, it utilizes a monitor lock, attached to the pipeline object. We can think of it as a flag. When a thread wants to enter the synchronized block, it must acquire this flag. If another thread already holds it, arriving threads are placed into a wait queue.

Java implements this "flag" using a counter: when a thread grabs the lock, the counter goes to 1. As soon as it's done, it sets back the counter to 0, and the waiting thread from the queue is allowed to proceed.

When working with threads we need to pay attention to

• Deadlocks: This occurs when Thread 1 waits for Thread 2, while Thread 2 waits for Thread 1. Both threads freeze forever.

• Starvation: This occurs when a thread is permanently denied access to shared resources because other high-priority threads keep stealing the execution slots.

Luckily our code does not suffer from these.

package concurrency;

import framework.Task;
import core.MathPipeline;
import java.util.List;

public class ConcurrentEngine {
    private final MathPipeline pipeline;

    public ConcurrentEngine(MathPipeline pipeline) {
        this.pipeline = pipeline;
    }

    public void startWorker(String workerName) {

        Thread thread = new Thread(() -> {
            System.out.println(workerName + " booted and ready.");

            while (true) {
                Task<Double> taskToExecute = null;

                synchronized (pipeline) {
                    taskToExecute = pipeline.pollTask();
                }

                if (taskToExecute != null) {
                    System.out.print("[" + workerName + "] ");
                    taskToExecute.execute();
                } else {
                    System.out.println(workerName + " finished all available work.");
                    break;
                }
            }
        });

        thread.setName(workerName);
        thread.start();
    }
}

Finally, we can implement out Main class. Here we will define a new pipeline, add 2 tasks and then execute them in parallel:

import core.*;
import concurrency.ConcurrentEngine;
import framework.Task;

public class Main {
    public static void main(String[] args) {
        MathPipeline pipeline = new MathPipeline();

        Task<Double> task1 = new SquareTask(4.0);
        Task<Double> task2 = new SquareTask(12.0);
        pipeline.addTask(task1);
        pipeline.addTask(task2);

        ConcurrentEngine engine = new ConcurrentEngine(pipeline);

        engine.startWorker("Thread-1");
        engine.startWorker("Thread-2");
    }
}

Building and Running

To run our Task Execution Pipeline, we open the terminal and navigate to the root directory of the project (where the src folder lives).

pipeline/
└── src/

Because our classes are organized into distinct packages and depend on one another, we use the -d flag with javac to specify a destination directory (we will name it bin) where the compiled .class bytecode files will be organized.

We can compile all files at once using wildcard matching:

javac -d bin src/framework/*.java src/core/*.java src/concurrency/*.java src/Main.java

Once successfully compiled, the bin directory will mirror our package structure.

To execute the application, we run the java launcher from the root folder, pointing it to the classpath folder (-cp bin) and specifying the class containing our main entry point:

java -cp bin Main

When executed, we will see our workers boot up on separate OS threads, request tasks from the synchronized pipeline and process them concurrently until the queue is safely emptied.