Developing reactive microservices

Microservices in a Reactive World

Up until now, the usual way to describe reactive applications has been to use a mix of technical and industry buzzwords, such as asynchronous, nonblocking, real-time, highly available, loosely coupled, scalable, fault-tolerant, concurrent, reactive, message-driven, push instead of pull, distributed, low latency, and high throughput.

The Reactive Manifesto brings all of these characteristics together and defines them through four high-level traits: responsive, resilient, elastic, and message-driven (Figure 2-1). Even if it looks like this describes an entirely new architectural pattern, the core principles have long been known in industries that require real-time IT systems, such as financial trading. If you think about systems composed out of individual services, you will realize how closely the reactive world is related to microservices. Let’s explore the four main characteristics of a reactive microservices system closer.

The Reactive Programming Model for Java Developers

But how do developers implement all of the above requirements and not use blocking and synchronous APIs? Reactive programming is the answer. At the very core this is a programming paradigm based on asynchronous message passing. This is used to establish a boundary between components that ensures loose coupling, isolation, and location transparency, and provides the means to delegate errors as messages.

You can create data streams of anything. Streams are cheap in terms of memory consumption and quite ubiquitous. You can use anything as a stream: variables, user inputs, properties, caches, and data structures. And if you’re thinking about StAX (streaming API for XML) right about now, I don’t blame you. Working on a stream of data and generating output while the final element hasn’t been received is exactly that kind of thinking—reacting to messages immediately when they arrive.

But reactive programming is closer to functional programming in general. A stream can be an input to another one. Even multiple streams can be used as inputs to another stream. You can merge two streams. You can filter a stream to get another one that has only those messages you are interested in. You can map data values from one stream to a new one. Java 8 introduced the Streams API, but this is a functional view over collections. Java 9 is slated to include the Reactive Streams specification.

Let’s look at a simple example. Let’s assume you have a bunch of receipts in your system and you need to find the ones you received for refueling your car ordered by receipt amount. Here’s the complete example before Java 8:

List<Receipt> receipts = new Arraylist<>();
for(Receipt r: receipts){
  if(r.getType() == Receipt.FUEL){
    fuelTransactions.add(r);
  }
}
Collections.sort(fuelTransactions, new Comparator(){
  public int compare(Receipt r1, Receipt r2){
    return r2.getAmount().compareTo(r1.getAmount());
  }
});
List<Integer> byFuelAmount = new ArrayList<>();
for(Receipt r: fuelTransactions){
  byFuelAmount.add(r.getId());
}

That’s a lot of code. And it is executed one step after another. When you work with the Streams API of Java 8 the same code looks like this:

List<Integer> byFuelAmount =
        receipts.parallelStream()
                .filter(r -> r.getType() == Receipt.FUEL)
                .sorted(comparing(Transaction::getAmount).
                 reversed())
                .map(Receipt::getId)
                .collect(toList());

Reactive programming raises the level of abstraction in your code so you can focus on the interdependence of messages that define the business logic, rather than having to handle a large amount of implementation details. And working with streams is only one part of it. In order to be fast and lightweight, the number of updates to existing data structures should be limited or eliminated. This is where immutable (or unmodifiable) data structures come in.

Basic Microservices Requirements

When designing individual reactive microservices, it is important to adhere to the core traits of:

• Isolation

• Autonomy

• Single responsibility

• Exclusive state

• Asynchronous message passing

• Mobility

Microservices are collaborative in nature and only make sense as systems. A complete explanation of all of these traits can be found in the Reactive Microservices Architecture report.

Java and Scala

The first version of Lagom provides a Java API for writing microservices. Before long, a subsequent version will add a Scala API.
Lagom’s Java APIs target Java 8. They assume familiarity with Java 8 features such as lambdas, default methods, and optional.

sbt Build Tool

One place where some light Scala coding is required, even for Java users, is in Lagom build definitions, which must be written using sbt’s Scala DSL. Lagom’s sbt-based development console allows you to run any number of services together with a single command.

Cassandra

Lagom has support for Cassandra as a data store, both for the write-side and read-side. Cassandra is a very scalable distributed database, and it is also flexible enough to support typical use cases of reactive services. By default, Lagom services needing to persist data use Cassandra as a database. For convenience, the development environment embeds a Cassandra server.

Play Framework

Lagom is implemented on top of the Play Framework. This is an implementation detail that will not directly concern simple microservices. More advanced users may wish to use some Play APIs directly. Play is built on the popular standard Netty network transport library. If you have an existing Play Framework application that you want to add microservices to there is an sbt plugin (PlayLagom) to help you.

Guice

Like Play, Lagom uses Guice for dependency injection.

Akka

Lagom Persistence, Publish-Subscribe, and Cluster are implemented on top of Akka, a toolkit for building concurrent, distributed, and resilient message-driven applications. This is an implementation detail that will not directly concern simple microservices. More advanced users may wish to use some Akka APIs directly.

Akka Streams

A Lagom service may be “simple” or “streamed”; this is described further later in this report in the section Service Implementation. Streaming asynchronous Lagom services are built on top of Akka Streams.

Akka Cluster

If you want to scale your microservices out across multiple servers, Lagom provides clustering via Akka Cluster.

Configuration

Lagom and many of its component technologies are configured using the Lightbend Config library. The configuration file format is HOCON, a powerful and expressive superset of JSON.

Logging

Lagom uses SLF4J for logging, backed by Logback as its default logging engine.

Example Application

The core concepts of reactive microservices have been introduced by Jonas Bonér in his report, Reactive Microservices Architecture. He clearly explains the main characteristics of microservices as a set of isolated services, each with a single area of responsibility. This forms the basis for being able to treat each service as a single unit that lives and dies in isolation—a prerequisite for resilience—and can be moved around in isolation—a prerequisite for elasticity. But they also own their state.

Each microservice has to take sole responsibility for its own state and persistence. Modeling each service as a bounded context can be helpful since each service usually defines its own domain, each with its own ubiquitous language. Both of these techniques are taken from domain-driven design (DDD). Of all the new concepts introduced in reactive microservices applications, consider DDD a good place to start learning. Microservices and the terms you hear in context are heavily influenced by DDD.

The philosophy of DDD is about placing the attention at the heart of the application, focusing on the complexity of the core business domain. Alongside the core business features, you’ll also find supporting subdomains that are often generic in nature, such as money or time. DDD aims to create models of a problem domain. All the implementation details—like persistence, user interfaces, and messaging—come later.

DDD was first introduced by Eric Evans in his book Domain-Driven Design: Tackling Complexity in the Heart of Software. In this book he also discusses an example application, the Cargo tracker. The example in this report follows the main business ideas showcased there and implements a first set of features.

Getting Started with the Example Application

The source code for the example application is available on GitHub. Please keep in mind that it is not a complete implementation of the well-known DDD example but merely a starting point. It will help you understand how to develop a reactive microservices system and is meant to be a playground for getting started.

The first step is to install sbt version 0.13.11. Make sure to have Homebrew installed:

$ brew install sbt

After you’re done with this you can fork and clone the example GitHub repository to your local machine:

$ git clone git@github.com:<username>/activator-lagom-
  cargotracker.git

If you want to contribute to the project, make sure to add the remote upstream accordingly. Change into the newly created directory and start sbt:

$ sbt
[info] Loading project definition from /cargotracker/project
[info] Set current project to cargotracker
(in build file:/cargotracker/)

You’ve now entered the sbt prompt. As a Java developer, keep in mind that sbt is not really comparable with Maven. sbt is a lot more than just a build tool. It is a complete development environment that offers an efficient, powerful, and convenient development experience.
A Lagom system is typically made up of a set of sbt builds, each build providing multiple services. If you inspect the example project’s directory structure, you’ll see the following:

• API and implementation projects for each service (shipping and registration)

• A front-end project that contains the ReactJS-based web-ui for the example

• The project folder with sbt-specific files.

• The build.sbt file, which contains all the information necessary to build, run, and deploy your services

Let’s take the example for a little test drive before diving in deeper:

$ sbt runAll

[info] Starting embedded Cassandra server
..........
[info] Cassandra server running at 127.0.0.1:4000
..........
[info] Service locator is running at http://localhost:8000
[info] Service gateway is running at http://localhost:9000
..........
[info] Service registration-impl listening for
       HTTP on 0:0:0:0:0:0:0:0:20019
[info] Service shipping-impl listening for
       HTTP on 0:0:0:0:0:0:0:0:25003
[info] Service front-end listening for
       HTTP on 0:0:0:0:0:0:0:0:28118
[info] (Services started, use Ctrl+D to stop and go
        back to the console...)

When inspecting the list of running services, you may wonder how ports are being assigned. Ports are assigned consistently, meaning that once a port is assigned to a service it will never change again. The port is deterministic even for different machines. An algorithm creates it out of the service and project name and maps it onto the default port range. With that said, you can change the settings easily.

Pointing your browser to the service gateway address will bring up the screen shown in Figure 3-1.

You can enter random text in each field and click “Post” to get it submitted. If you check the logs you’ll see that a cargo with an assigned ID was persisted:

[info] s.c.r.i.RegistrationServiceImpl - Cargo ID: 154038.

Now it’s time to look at what happened under the hood. Open the project in IntelliJ and review the registration service. You can also refer to the Lagom documentation for importing your project into your IDE of choice.

API Versus Implementation

There are two projects with the prefix registration. One is the API project and the other the implementation project. The API project contains a service interface through which consumers may interact with the service, while the implementation project contains the actual service implementation.

The registration-api project contains two classes. The RegistrationService contains the service description that is defined by an interface. This interface not only defines how the service is invoked and implemented, it also defines the metadata that describes how the interface is mapped down onto an underlying transport protocol. Generally, the service descriptor, and its implementation and consumption, should remain agnostic to what transport is being used, regardless of whether that’s REST, Websocket, or other transports.

import akka.NotUsed;
import com.lightbend.lagom.javadsl.api.*;

public interface RegistrationService extends Service {

   ServiceCall<NotUsed, Cargo, Done> register();
   ServiceCall<NotUsed, NotUsed, Source<Cargo, ?>>
                             getLiveRegistrations();

   @Override
   default Descriptor descriptor() {
       return named("registrationService").with(
           restCall(Method.POST, "/api/registration",
                    register()),
           pathCall("/api/registration/live",
                    getLiveRegistrations())
         ).withAutoAcl(true);
   }
}

The descriptor defines a service with two calls: register() and getLiveRegistrations(). Both return a ServiceCall, which is a representation of the call that can be invoked when consuming the service. Unlike what most Java developers are used to, an invocation of both methods does not actually invoke the call but gives you a handle to it, which has to be invoked with the ServiceCall.invoke() method. A ServiceCall takes three type parameters and id, which is extracted from the incoming identifier, e.g., the path in the case of a REST request. As the akka.NotUsed indicates, this example doesn’t have an ID and its REST transport implementation is going to use a static path. Request and Response reflect incoming and outgoing messages. Cargo is the request and akka.Done the response.

After implementing and invoking the register() call, the mapping to the actual transport is still missing. This is done by providing a default implementation of the Service.descriptor() method. The implementation is pretty self-explanatory. A service named registrationService is returned that contains two REST calls. The call mapping to the register() service call is a POST request to /api/registration and the getLiveRegistrations() call is a GET request mapped to /api/registration/live.

We have to look back to the identifiers a bit. In fact, the mapping to the transport is only one aspect. Even more important is that each service call needs to have an identifier to provide routing information to the client and the service to make sure the calls are mapped appropriately. As explained before, our example doesn’t use a dynamic part and thus returns akka.NotUsed. But we still need one and this example uses both, a path-based identifier that uses a URI path and query string (pathCall()) to route calls and a REST identifier (restCall()). REST identifiers should be used when you’re creating a semantic REST API. They use both a path, as with the path-based identifier, and a request method to identify them. To learn more about configuration and usage of identifiers in general, refer to the path-based identifier section in the Lagom documentation.

We already learned that reactive microservices systems communicate via messages. Every service call in Lagom has a request message type and a response message type that fall into two categories:

• A strict message is a single message that can be represented by a simple Java object. The message will be buffered into memory, and then parsed, for example, as JSON. When both message types are strict, the call is said to be a synchronous call, i.e., a request is sent and received, then a response is sent and received. The caller and callee have synchronized their communication.

• A streamed message is a message of type Source, which is an Akka Streams API that allows asynchronous streaming and handling of messages. This is what the getLiveRegistrations() call returns. A stream of Source<Cargo> is returned as the response message. Instead of using synchronous REST calls, Lagom chooses an appropriate transport for the stream, which will typically be WebSockets. WebSockets support bidirectional streaming, and is a good general-purpose option for streaming.

By default all messages are serialized and deserialized to JSON by using Jackson. To change this behavior or to write and configure custom message serializers, see Message Serializers in the Lagom documentation.

The message passed around in this example is a Cargo object. And it is also part of the registration-api project because it is part of the external API.

import
  com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import com.lightbend.lagom.javadsl.immutable.ImmutableStyle;
import org.immutables.value.Value;

@Value.Immutable
@ImmutableStyle
@JsonDeserialize(as = Cargo.class)
public interface AbstractCargo {

    @Value.Parameter
    String getId();

    @Value.Parameter
    String getName();

    //...
}

This is a simple value object on first sight. But looking closer, we can see that it is an immutable object, which means it cannot be modified after it is created. All fields are final and are assigned at construction time. There are no setter methods.

Immutable objects have two great advantages:

• Code based on immutable objects is clearer and likelier to be correct. Bugs involving unexpected changes simply can’t occur.

• Multiple threads can safely access immutable objects concurrently.

The generated Cargo.java class can be found in the target/scala-2.11/src_managed folder of the registration-api project. Deeper coverage about immutables and how to use them is part of the Lagom documentation.

We have now explored the complete registration-api project and it is time to move on to the service implementation.

Service Implementation

Services are implemented by providing an implementation of the service descriptor interface and all its calls. Before we dive into the implementation we need to make sure it is registered with Lagom. The service implementation is a module and the registration can be done by extending AbstractModule and implementing ServiceGuiceSupport:

public class RegistrationServiceModule extends
             AbstractModule
implements ServiceGuiceSupport {
    @Override
    protected void configure() {
      bindServices(serviceBinding(RegistrationService.class,
      RegistrationServiceImpl.class));
    }
}

Now onto the actual service, RegistrationServiceImpl. The first call we tried out as a smoke test was register(). And now that you have seen the service descriptor, we can look at the implementation:

private final PersistentEntityRegistry persistentEntityRegistry;
private final PubSubRegistry topics;
private final Logger log =
         LoggerFactory.getLogger(RegistrationServiceImpl.class);

@Override
   public ServiceCall<NotUsed, Cargo, Done> register() {
       return (id, request) -> {
           // Publish received entity into topic named "Topic"
           PubSubRef<Cargo> topic = topics.refFor(
                            TopicId.of(Cargo.class, "topic"));
           topic.publish(request);

           log.info("Cargo ID: {}.", request.getId());

          //...
       };
   }

The service call just returns akka.Done. Let’s try to find out what actually happens here. The first two lines represent a publish/subscribe topic. The line starting with PubSubRef<Cargo> topic = ... creates a topic with the name “topic” and registers it with the PubSubRegistry. The next line publishes a single object into the topic. You can find the consuming side of the topic in the second service call, getLiveRegistrations().

A simple log entry via Log4J was the initial smoke test. And now it is time to learn about how to persist data.

Consuming Services

We’ve seen how to define service descriptors and implement them, but now we need to consume services. The service descriptor contains everything Lagom needs to know to invoke a service. Consequently, Lagom is able to implement service descriptor interfaces for you.

The first thing necessary to consume a service is to bind it, so that Lagom can provide an implementation for your application to use. We’ve done that with the service before. Let’s add a client call from the shipping-impl to the registration-api and validate a piece of cargo before we add a leg in the shipping-impl:

public class ShippingServiceModule extends
      AbstractModule implements ServiceGuiceSupport {
    @Override
    protected void configure() {
        bindServices(serviceBinding(ShippingService.class,
        ShippingServiceImpl.class));
        bindClient(RegistrationService.class);
    }
}

Make sure to also add the dependency between both projects in the build.sbt file by adding .dependsOn(registrationApi) to the shipping-impl project.

Having bound the client, you can now have it injected into any Lagom component using the @Inject annotation. In this example it is injected into the ShippingServiceImpl:

public class ShippingServiceImpl implements ShippingService {
    private final RegistrationService registrationService;

    @Inject
    public ShippingServiceImpl(PersistentEntityRegistry
    persistentEntityRegistry,
    RegistrationService registrationService) {
        this.registrationService = registrationService;
        //...
    }

The service can be used to validate a cargo ID in the shipping-impl before adding a leg:

@Override
    public ServiceCall<String, Leg, Done> addLeg() {
        return (id, request) -> {
            CompletionStage<Cargo> response =
            registrationService.getRegistration()
              .invoke(request.getCargoId(),
                 NotUsed.getInstance());
            PersistentEntityRef<ShippingCommand>
            itinerary = persistentEntityRegistry
                    .refFor(ItineraryEntity.class, id);
            return itinerary.ask(AddLeg.of(request));
        };
    }

All service calls with Lagom service clients are by default using circuit breakers. Circuit breakers are used and configured on the client side, but the granularity and configuration identifiers are defined by the service provider. By default, one circuit breaker instance is used for all calls (methods) to another service. It is possible to set a unique circuit breaker identifier for each method to use a separate circuit breaker instance for each method. It is also possible to group related methods by using the same identifier on several methods. You can find more information about how to configure the circuit breaker in the Lagom documentation.

Migration Approaches

While Lagom and the reactive programing model is clearly favoring the greenfield approach, nothing is stopping you from striving for a brownfield migration. You have three different ways to get started with this.

Legacy Integration

Even if the term “legacy” has an old and outdated touch to it, I use it to describe everything that is not already in a microservices-based architecture. You have existing business logic in your own applications. There are libraries and frameworks that provide access to some proprietary system and there may be host systems that need to be integrated. More specifically, this is everything that exists and still needs to function while you’re starting to modernize your applications. There are many ways to successfully do this. Please keep in mind, though, that this isn’t an architectural discussion of enterprise integration but a technical assessment of the interface technologies and how you can use them. Figure 5-1 gives a high-level overview about integration technologies and how to use them with Lagom.

Lagom builds upon and integrates very tightly with both Akka and Play. You can use Akka from your Lagom service implementations by injecting the current ActorSystem into it or directly into persistent entities with ordinary dependency injection. Details about the integration can be found in the Lagom documentation.