Commands

Every system issues commands. Commands are actions that a user wishes to perform. Most HTTP-based APIs are passing commands: the client asks for an action to happen. It’s important to understand that the action has not yet happened. It may happen in the future, or not, it may complete successfully or fail. In general commands are sent to a specific recipient, and a result is sent back to the client.

Take the simple HTTP application we used in Chapter 3. You emitted a simple HTTP request. As said above, that was a command. The application receives that command, handles it, and produces a result.

Events

Events are actions that have successfully completed. An event represents a fact, something that happened: a keystroke, a failure, an order, anything important to the organization or system at hand that happened. An event can be the result of work done by a command.

Let’s go back to the HTTP request example from above. Once the response has been written, it had become an event. We have seen an HTTP request and its response. That event can be written in a log, or broadcast to interested parties can be aware of what happened.

Events are immutable. You cannot delete an event. Admittedly, you can’t change the past. If you want to refute a previously sent fact, you need to fire another event invalidating the fact. The carried facts are only made irrelevant by another fact for the current knowledge.

Messages

But, how to publish these events? There are many ways to publish events. These days, solutions like Apache Kafka or Apache ActiveMQ (we will cover both in Chapter 11) are very popular. They act as brokers between the producers and consumers. Essentially, our events are written into topics or queues. To write these events, the application sends a message to the broker targeting a specific destination (the queue or the topic).

A message is a self-contained data structure describing the event and any relevant details about the event such as who emitted it, at what time, potentially a unique ID. It’s generally better to keep the event itself business-centric and use additional metadata for the technical aspects.

On the other side, to consume events, you subscribe to the queue or topic containing the events you are interested in, and receive the messages. You unwrap the event and can also get the associated metadata (like when did the event happened, where did it happen…). The processing of an event can lead to the publication to other events (again packaged in messages and sent to a known destination), or to the execution of commands.

Brokers and messages can also convey commands. In this case the message contains the description of the action to execute, and another message (potentially multiple of them) would carry the outcome if needed.

Commands vs. Events: an example

Let’s take a look at an example to highlight the differences between commands and events. Imagine an e-commerce shop, as the one depicted on Figure 4-2. The user picks a set of products and finalizes the order (process to payment, get the delivery date, etc.).

Figure 4-2. Simplified architecture of an e-commerce shop

The user sends a command, using an HTTP request for example, to the shop service with the items he wishes to receive. In a traditional application, once the ShopService receives the command, it would call an OrderService and invoke an order method with the username, the list of items (basket), and so on. Calling the order method is a command. That makes the ShopService dependant of the OrderService and reduce the component autonomy: the ShopService cannot operate without the OrderService. We are creating a distributed monolith¹, i.e. a distributed application that would collapse as soon as one of its parts fails.

Let’s see the difference if, instead of using a command between the ShopService and OrderService, we publish an event. Once the user finalizes the order, it still sends a command to the shop service. However, this time, the shop service transforms that command into an event: a new order has been placed. The event contains the user, the basket, and so on. It’s a fact written in a log, or wrapped into a message and sent to a broker.

On the other side, the OrderService observes the a new order has been placed events, by reading where these events are stored. When the shop service emits the event, it receives it and can process it.

With this architecture, the ShopService does not depend on the OrderService. In addition, the OrderService does not depend on the ShopService, and it would process any observed event, regardless the emitter. For example, a mobile application can emit the same event when the user validates an order from a mobile phone.

Multiple components can consume events (Figure 4-3). For example, in addition to the OrderService, the StatisticsService keeps track of the most ordered items. It consumes the same event, without having to modify the ShopService to receive them.

A component observing events can derive new ones from them. For instance, the StatisticsService could analyze the order and compute recommendations. These recommendations could be seen as another fact, and so communicate as an event. The ShopService could observe these events and process them to influence items selection. However, the StatisticsService and the ShopService are independent of each other. The knowledge is cumulative and occurs by receiving new events and deriving, as done by the StatisticsService, new facts from the received events.

Figure 4-3. Architecture of the e-commerce shop with events and message queues

As depicted on the image in Figure 4-3, we can use message queues to transport our event. These events are wrapped into messages, sent to known destination (orders and recommendations). The OrderService and StatisticService consume and process the messages independently.

It’s important for these destinations persist the events as an ordered sequence. By keeping that sequence, the system can go back in time, and reprocess the events. Such replay mechanism, very popular in the Kafka world, has multiple benefits. You can restart with a clean state after a disaster by reprocessing all the stored events. Then, if for example, we change the recommendation algorithm from the statistic service, it would be able to re-accumulate all the knowledge and derive new recommendations.

While in this example, the event emission sounds explicit, it’s not always the case. For example, events can be created from database writes².

Commands and events are the basis of most of the interactions. While we use mostly commands, events come with significant benefits. Events are facts. Events tell a story, the story of your system, a narrative that describes your system’s evolution.

Figure 4-4. Overview of a reactive system

In reactive systems, events are wrapped into messages, and these messages are sent to destination, transported by message broker such as AMQP or Kafka (Figure 4-4). Such an approach solves two important architectural issues arising from the distributed systems. First, it naturally handles real-world asynchronicity. Second, it binds together services without relying on strong-coupling. At the edge of the system, it uses commands most of the time, often relying on HTTP.

This asynchronous message passing aspect of reactive systems forms the communication tissue. It not only grants the applications forming the system more autonomy and independence, but it also enables resilience and elasticity. You may wonder how? You will get the beginning of the response in the next section.

Blocking Network I/O, threads and concurrency

To understand the benefits of non-blocking I/O, we need to know how blocking I/Os work. Let’s use a client-server interaction to illustrate this. When a client sends a request to a server, the server processes it and sends back a response. HTTP, for instance, follows this principle. For this to happen, both the client and the server need to establish a connection before the interaction starts. We will not go into the depths of the 7-layers model and the protocol stack involved in this interaction; you can find plenty of articles online about that topic.

So, to establish that connection between the client and the server, we use sockets:

int port = 9999;

// Create a server socket
try (ServerSocket server = new ServerSocket(port)) {
    while (true) {

        // Wait for the next connection from a client
        Socket client = server.accept();

        PrintWriter response = new PrintWriter(client.getOutputStream(), true);
        BufferedReader request = new BufferedReader(
                new InputStreamReader(client.getInputStream()));

        String line;
        while ((line = request.readLine()) != null) {
            System.out.println("Server received message from client: " + line);
            // Echo the request
            response.println(line);

            // Add a way to stop the application.
            if ("done".equalsIgnoreCase(line)) {
                break;
            }
        }
        client.close();
    }
}

The client and the server have to bind themselves to a socket forming the connection. The server listens to its socket for the client to connect. Once established, the client and the server can both write and read data from the socket bound to that connection.

Traditionally, because it’s simpler, applications are developed using a synchronous development model. Such a development model executes instructions sequentially, one after the other. So, when such applications interact across the network, they expect to continue using a synchronous development model even for I/O. It uses synchronous communication and blocks the execution until the operation completes. For example, in the previous code snippet (Example 4-1), we wait for a connection and handle it synchronously. We read and write using synchronous APIs. It’s simpler, but it leads to the usage of blocking I/O.

With the blocking I/O, when the client sends a request to the server, the socket processing that connection and the corresponding thread that reads from it is blocked until some read data appears. The bytes are accumulated in the network buffer until everything is read and ready for processing. Until the operation is complete, the server can do nothing more but wait.

The consequence of this model is that we cannot serve more than one connection within a single thread. When the server receives a connection, it uses that thread to read the request, process it, and write the response. That thread is blocked until the last byte of the response is written on the wire. A single client connection blocks the server! Not very efficient, right?

So, to execute concurrent requests with this approach, the only way is to have multiple threads. So we need to allocate a new thread for each client connection To handle more clients, you need more threads and process each request on different worker thread:

while (listening) {
    accept a connection;
    create a worker thread to process the client request;
}

To implement this principle, we need a thread pool (worker pool). When the client connects, we accept the connection and off-load the processing to a separate thread. Thus, the server thread can still accept other connections:

int port = 9999;
ExecutorService executors = Executors.newFixedThreadPool(10); 

// Create a server socket
try (ServerSocket server = new ServerSocket(port)) {
    while (true) {

        // Wait for the next connection from a client
        Socket client = server.accept();

        executors.submit(() -> {                                    
            try {
                PrintWriter response = new PrintWriter(client.getOutputStream(), true);
                BufferedReader request = new BufferedReader(
                        new InputStreamReader(client.getInputStream()));

                String line;
                while ((line = request.readLine()) != null) {
                    System.out.println(Thread.currentThread().getName() +
                            " - Server received message from client: " + line);
                    // Echo the request
                    response.println(line);

                    // Add a way to stop the application.
                    if ("done".equalsIgnoreCase(line)) {
                        break;
                    }
                }
                client.close();
            } catch (Exception e) {
                System.err.println("Couldn't serve I/O: " + e.toString());

            }
        });
    }
}

That’s the model used, by default, in traditional Java frameworks such as Jakarta or Spring³. Even if under the hood, these frameworks may use non-blocking I/O, it uses worker threads to handle the requests. But, there are many drawbacks to this approach:

1. Each thread requires a stack of memory allocated to it. With the increasing number of connections, spawning multiple threads and switching between them will consume not only memory but also CPU cycles.

2. At any given point in time, there can be multiple threads just waiting for the client requests. That’s a massive waste of resources.

3. Your concurrency (the number of requests you can handle at a given time - 10 in the previous example) is limited by the number of threads you can create.

On public Clouds, it inflates your monthly bill, on private clouds, it reduces the deployment density. Therefore, the blocking I/O approach is not ideal if you have to handle many connections or implement applications dealing with a lot of I/O. In the realm of distributed systems, that’s often the case.

Luckily, there’s an alternative.

How non-blocking I/O work?

The alternative is non-blocking I/O. The difference is evident from its name. Instead of blocking waiting for the completion of the transmission, the caller is not blocked and can continue its processing. The magic happens in the operating system. With non-blocking I/O, the operating system queues the requests. The system processes the actual I/O in the future. When the I/O completes, and the response is ready, a continuation, often implemented as a callback, happens and the caller receives the result.

To better understand the benefits and see how these continuations work, we need to look under the hood: how are non-blocking I/O implemented. We already mentioned a queue. It enqueues I/O operations and returns immediately, so the caller is not blocked waiting for the I/O operations to complete. When a response comes back, the system stores the result in a structure. When the caller needs the result, it interrogates the system to see if the operation completed:

InetSocketAddress address = new InetSocketAddress("localhost", 9999);
Selector selector = Selector.open();
ServerSocketChannel channel = ServerSocketChannel.open();
channel.configureBlocking(false);

channel.socket().bind(address);
// Server socket only support ACCEPT
channel.register(selector, SelectionKey.OP_ACCEPT);

while (true) {
    int available = selector.select(); // wait for events
    if (available == 0) {
        continue;  // Nothing ready yet.
    }

    // We got request ready to be processed.
    Set<SelectionKey> keys = selector.selectedKeys();
    Iterator<SelectionKey> iterator = keys.iterator();
    while (iterator.hasNext()) {
        SelectionKey key = iterator.next();
        if (key.isAcceptable()) {
            // --  New connection --
            SocketChannel client = channel.accept();
            client.configureBlocking(false);
            client.register(selector, SelectionKey.OP_READ);
            System.out.println("Client connection accepted: " + client.getLocalAddress());
        } else if (key.isReadable()) {
            // --  A client sent data ready to be read and we can write --
            SocketChannel client = (SocketChannel) key.channel();
            // Read the data assuming the size is sufficient for reading.
            ByteBuffer payload = ByteBuffer.allocate(256);
            int size = client.read(payload);
            if (size == -1 ) { // Handle disconnection
                System.out.println("Disconnection from " + client.getRemoteAddress());
                channel.close();
                key.cancel();
            } else {
                String result = new String(payload.array(), StandardCharsets.UTF_8).trim();
                System.out.println("Received message: " + result);
                if (result.equals("done")) {
                    client.close();
                }
                payload.rewind(); // Echo
                client.write(payload);
            }
        }
        // We sure we don't handle it twice.
        iterator.remove();
    }
}

Non-blocking I/O introduces a few new concepts:

• we don’t use InputStream or OutputStream (which are blocking by nature), but Buffer. A Buffer is a temporary storage.

• Channel’s can be viewed as an endpoint for an open connection.

• Selector is the cornerstone of non-blocking I/O in Java.

A Selector manages multiple channels, either server or client channels. When you use non-blocking I/O, you create a Selector. Each time you deal with a new channel, you register this channel on the selector with the events your are interested in (accept, ready to read, ready to write).

Then, your code polls the Selector with only one thread to see if the channel is ready. When the channel is ready to read or write, you can start to read and write. We don’t need to have a thread for every channel at all, and a single thread can handle many channels.

The selector is an abstraction of the non-blocking I/O implementation provided by the underlying operating system. Various approaches depending on the operating systems are available.

First, select was implemented in the 1980s. It supports the registration of 1024 sockets. That was certainly enough in the 80’s, but not anymore.

poll is a replacement for select introduced in 1997. The most significant difference is that poll no longer limits the number of sockets. However, as select, the system only tells you how many channels are ready, but not which ones. So, you need to iterate over the set of channels to check which ones are ready. When there are few channels, it is not a big problem. Once the number of channels is more than hundreds of thousands, the iteration time is considerable.

Then, epoll appeared in 2002 in the Linux Kernel 2.5.44. Kqueue appeared in FreeBSD in 2000 and /dev/poll in Solaris around the same time. These mechanisms return the set of channels that are ready to be processed - no more iteration over every channel! Finally, windows systems provide IOCP, a very optimized implementation of select.

What’s important to remember is that regardless of how the operating systems implement it, with non-blocking I/O, you only need a single thread to handle multiple requests. It makes the model much more efficient than blocking I/O, as you don’t need to create threads to handle concurrent requests. Eliminating these extra threads makes your application much more efficient in terms of memory consumption (~ 1Mb per thread) and avoids wasting CPU cycles due to context switches (1-2 us per switch)⁴.

Reactive systems recommend the usage of non-blocking I/O to receive and send messages. Thus, your application can handle more messages with fewer resources. Another advantage is that an idle application would consume almost no memory or CPUs. You don’t have to reserve resources upfront.

Reactor pattern and event loop

Non-Blocking I/O gives us the possibility to handle multiple concurrent requests or messages with a single thread. How could we handle these concurrent requests? How do we structure our code when using non-blocking I/O? The examples given in the previous section are not scaling well; we can quickly see that implementing a REST API with such a model will be a nightmare. Besides, we would like to avoid using worker threads, as it would discard the advantages of non-blocking I/O. So we need something different: the reactor pattern.

The reactor pattern allows associating I/O events with event handlers. The reactor, the cornerstone of this mechanism, invokes the event handlers when the expected event is received.

The purpose of the reactor pattern is to avoid creating a thread for each message, request, and connection. It receives events from multiple channels and sequentially distributes them to the corresponding event handlers.

Figure 4-7. The reactor pattern

Implementation of the reactor pattern uses an event loop (Figure 4-7). It’s a thread iterating over the set of channels, and when there is some data ready to be consumed, invokes the associated event handler sequentially, in a single-threaded manner.

When you combine non-blocking I/O and the reactor patterns, you organize your code as a set of event handlers. That approach works wonderfully with reactive code as it exposes the notion of events, the essence of Reactive.

There are a few variants of the reactor pattern: * The multi-reactor pattern uses multiple event loops (generally 1 or 2 per CPU core), which increase the concurrency of the application. Multi-reactor pattern implementations, such as Eclipse Vert.x, call the event handlers in a single-threaded manner to avoid deadlock or state visibility issues. * The proactor pattern can be seen as an asynchronous version of the reactor pattern. With this proactor pattern, long-running event handlers invoke a continuation when they complete. Such mechanism allow mixing non-blocking and blocking (Figure 4-8).

Figure 4-8. The proactor pattern

You can integrate non-blocking event handlers as well as blocking ones, by offloading their execution to separated threads when it’s inevitable. When their execution completes, it invokes the continuation. As we will see in Chapter 6, this is the pattern used by Quarkus.