Microservices with communication via Axon

In this simple tutorial, we will create a couple of microservices using Spring Boot and organize their interaction through the Axon framework.

Let's assume we have the following task.

There is a source of trades in the stock market. This source sends us trades via a REST interface.

We need to obtain these trades, store them in a database, and create a convenient in-memory storage.

This storage must perform the following functions:

• return a list of trades;
• return the full position, i.e., a table of "instrument" — "current quantity of papers";
• return the position for a given instrument.

How will we approach solving this task?

Following the principles of microservice architecture, we need to divide the task into component microservices:

• fetch the trade via REST;
• save the trade to the database;
• an in-memory storage for data representation of the position.

In this tutorial, let's implement the first and third services, leaving the second for the next part (let us know in the comments if you're interested).

So, we have two microservices.

The first one receives data from the outside.

The second processes this data and responds to incoming requests.

Of course, we want to achieve horizontal scalability, zero-downtime updates, and other advantages of microservices.

What, quite a challenging, task lies ahead of us?

In fact, there are many, but right now, let's discuss how data will flow between these microservices. We can set up REST between them, implement a queue, or come up with various options, each with its own pros and cons.

Let's consider one possible approach – asynchronous interaction via the Axon framework..

What are the advantages of this solution?

Firstly, asynchronous interaction increases flexibility (yes, there is a downside, but for now, we will focus on the positives).

Secondly, out of the box, we get Event Sourcing and CQRS.
Thirdly, Axon provides a ready-made infrastructure, allowing us to concentrate solely on developing business logic.

Let's get started.

Our project will be based on Gradle. It will consist of three modules:

• common. a module with shared data structures (we dislike copy-pasting);
• tradeCreator. a module with a microservice for receiving trades via REST;
• tradeQueries. a module with a microservice for displaying the position.

Let's take Spring Boot as the foundation and connect the Axon starter.

Axon works perfectly even without Spring, but we will use them together.

Here we need to pause and say a few words about Axon.

It is a client-server system. There is a server – a separate application that we will run in Docker.

And there are clients that are embedded in microservices.
This is what the picture looks like. First, the Axon server (in Docker) is started, then our microservices.

When starting, microservices look for the server and begin to interact with it. This interaction can be conditionally divided into two types: technical and business.

Technical is the exchange of messages like "I'm alive" (such messages can be seen in debug logging mode).

Business is the exchange of messages like "new deal."

An important feature is that after starting, a microservice can ask the Axon server "what happened," and the server sends the accumulated events to the microservice. This way, the microservice can be restarted relatively safely without data loss.
With this exchange scheme, we can easily start many instances of microservices,
even on different hosts.

Yes, having a single instance of the Axon server is not reliable, but for now, that's how it is.

We work within the paradigms of Event Sourcing and CQRS. This means we should have "commands," "events," and "queries."

We will have one command: "create deal," one event: "deal created," and three queries: "show all deals," "show position," "show position by instrument."

The workflow looks like this:

1. The tradeCreator microservice receives a deal via REST.
2. The tradeCreator microservice creates the command "create deal" and sends it to the Axon server.
3. The Axon server receives the command and forwards it to the interested recipient, in our case, this is the tradeCreator microservice.
4. The tradeCreator microservice receives the command, creates the event "deal created," and sends it to the Axon server.
5. The Axon server receives the event and forwards it to interested subscribers.
6. Currently, there is only one interested recipient – the tradeQueries microservice.
7. The tradeQueries microservice receives the event and updates its internal data.

(It’s important that at the moment of the event creation, the tradeQueries microservice may be unavailable, but as soon as it starts, it will immediately receive the event).

Yes, the Axon server is at the center of communications, all messages go through it.

Let's move on to coding.

To avoid cluttering the post with code, I will provide only snippets below, and a link to the complete example will be included later.

Let's start with the common module.

In it, the common parts include an event (class CreatedTradeEvent). Note the naming; essentially, this is the name of the command that generated this event, but in the past tense. It's in the past because the command emerges first, leading to the creation of the event.

Other common structures include classes for describing the position (class Position), the trade (class Trade), and the trade side (enum Side), meaning buying or selling.

Now let's move on to the tradeCreator module.

This module has a Rest interface (class TradeController) for accepting trades.
From the received trade, a command to 'create a trade' is formed and sent to the axon server.

    @PostMapping("/trade")
    public ResponseEntity create(@RequestBody Trade trade) {
        var createTradeCommand = CreateTradeCommand.builder()
                .tradeId(trade.getTradeId())
	...
                .build();
        var result = commandGateway.sendAndWait(createTradeCommand, 3, TimeUnit.SECONDS);
        return ResponseEntity.ok(result.get().toString());
    }

The command is processed using the class class TradeAggregate.
To let Axon find it, we apply the @Aggregate annotation.
The method for handling the command looks like this (with some parts shortened):

    @CommandHandler
    public TradeAggregate(CreateTradeCommand command) {
        log.info("command: {}", command);
        var event = CreatedTradeEvent.builder()
                .tradeId(command.tradeId())
		....
                .build();
        AggregateLifecycle.apply(event);
    }

An event is generated from the command and sent to the server.
The command is located in the class CreateTradeCommand.

Now let's look at the last module, tradeQueries.

Queries are described in the queries package.
This module also has a Rest interface.
public class TradeController.

For example, let's look at the handling of the request: 'show all trades'.

    @GetMapping("/trade/all")
    public List findAllTrades() {
        return queryGateway.query(new FindAllTradesQuery(),
                ResponseTypes.multipleInstancesOf(Trade.class)).join();
    }

A request for retrieval is created and sent to the server.

The class TradesEventHandler is used to process the retrieval request.
It contains a method marked with the annotation.

   @QueryHandler
    public List handleFindCurrentPositionQuery(FindCurrentPositionQuery query)

This method is responsible for fetching data from the in-memory storage.

The question arises: how is the information updated in this storage?

To start, it’s simply a set of ConcurrentHashMaps tailored for specific queries.
The method is used to update them:

    @EventHandler
    public void on(CreatedTradeEvent event) {
        log.info("event:{}", event);

        var trade = Trade.builder()
	...
                .build();
        trades.put(event.tradeId(), trade);
        position.merge(event.shortName(), event.size(),
                (oldValue, value) -> event.side() == Side.BUY ? oldValue + value : oldValue - value);
    }

It handles the "trade created" event and updates the Maps.

These are the key points in microservices development.

What can be said about the disadvantages of Axon?

First of all, it complicates the infrastructure, creating a single point of failure - the Axon server, with all communications routed through it.

Secondly, a significant drawback of such distributed systems arises - temporal inconsistency of data. In our case, an unacceptable amount of time may pass between receiving a new trade and updating the data for queries.

What was left unsaid?

There has been no mention of Event Sourcing and CQRS, what they are, and why they are needed.
Without explaining these concepts, some points may not be clear.

Perhaps some sections of code also require clarification.

We will discuss this at the open webinar on September 21.

Full example.

Source: habr.com