How to Implement CQRS (Command Query Responsibility Segregation) in Java

Introduction

Command Query Responsibility Segregation (CQRS) is a design pattern that separates the read and write operations of a data store. The main idea is to use different models to update data (command) and read data (query). This separation can provide several benefits, such as improved performance, scalability, and security. In this tutorial, we will explore how to implement CQRS in a Java application.

1. Understanding CQRS

What is CQRS?

CQRS stands for Command Query Responsibility Segregation. It’s a pattern that separates the responsibility of reading and writing data. In traditional CRUD (Create, Read, Update, Delete) applications, the same model is used for both reads and writes. CQRS, on the other hand, uses distinct models for commands (writes) and queries (reads).

Benefits of CQRS

• Scalability: By separating reads and writes, each can be scaled independently.
• Performance: Optimized read models can be designed to improve query performance.
• Security: Different models allow for fine-grained security policies.
• Maintainability: Simplifies the system by separating concerns.
• Event Sourcing: Often used with CQRS, providing an audit log of all changes.

CQRS vs. CRUD

• CRUD: Single model for both reading and writing. Simpler, but can become complex and less performant as the application grows.
• CQRS: Separate models for commands and queries. More complex initially but offers better scalability and performance.

2. Setting Up the Project

Project Structure

We’ll create a Maven project with the following structure:

cqrs-demo
├── src
│   ├── main
│   │   ├── java
│   │   │   ├── com
│   │   │   │   ├── example
│   │   │   │   │   ├── command
│   │   │   │   │   ├── query
│   │   │   │   │   ├── event
│   │   │   │   │   ├── model
│   │   │   │   │   ├── service
│   │   │   │   │   ├── bus
│   │   │   ├── resources
│   ├── testCode language: Bash (bash)

Dependencies

Add the following dependencies to your pom.xml file:

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>Code language: HTML, XML (xml)

3. Implementing the Write Model (Commands)

Command Model

Commands represent actions that change the state of the system. Each command should be immutable and only contain the data needed for the operation.

Create a command for creating a new user:

package com.example.command;

public class CreateUserCommand {
    private final String userId;
    private final String username;
    private final String email;

    public CreateUserCommand(String userId, String username, String email) {
        this.userId = userId;
        this.username = username;
        this.email = email;
    }

    public String getUserId() {
        return userId;
    }

    public String getUsername() {
        return username;
    }

    public String getEmail() {
        return email;
    }
}Code language: Java (java)

Command Handler

The command handler processes the commands. It should contain the business logic for handling each command.

package com.example.command;

import com.example.model.User;
import com.example.service.UserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class CreateUserCommandHandler {

    private final UserService userService;

    @Autowired
    public CreateUserCommandHandler(UserService userService) {
        this.userService = userService;
    }

    public void handle(CreateUserCommand command) {
        User user = new User(command.getUserId(), command.getUsername(), command.getEmail());
        userService.save(user);
    }
}Code language: Java (java)

Command Bus

The command bus is responsible for dispatching commands to their respective handlers.

package com.example.bus;

import com.example.command.CreateUserCommand;
import com.example.command.CreateUserCommandHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class CommandBus {

    private final CreateUserCommandHandler createUserCommandHandler;

    @Autowired
    public CommandBus(CreateUserCommandHandler createUserCommandHandler) {
        this.createUserCommandHandler = createUserCommandHandler;
    }

    public void dispatch(CreateUserCommand command) {
        createUserCommandHandler.handle(command);
    }
}Code language: Java (java)

4. Implementing the Read Model (Queries)

Query Model

Queries represent requests for data. Like commands, queries should be immutable.

Create a query to fetch user details:

package com.example.query;

public class GetUserQuery {
    private final String userId;

    public GetUserQuery(String userId) {
        this.userId = userId;
    }

    public String getUserId() {
        return userId;
    }
}Code language: Java (java)

Query Handler

The query handler processes the queries and returns the required data.

package com.example.query;

import com.example.model.User;
import com.example.service.UserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class GetUserQueryHandler {

    private final UserService userService;

    @Autowired
    public GetUserQueryHandler(UserService userService) {
        this.userService = userService;
    }

    public User handle(GetUserQuery query) {
        return userService.findById(query.getUserId());
    }
}Code language: Java (java)

Query Bus

The query bus is responsible for dispatching queries to their respective handlers.

package com.example.bus;

import com.example.query.GetUserQuery;
import com.example.query.GetUserQueryHandler;
import com.example.model.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class QueryBus {

    private final GetUserQueryHandler getUserQueryHandler;

    @Autowired
    public QueryBus(GetUserQueryHandler getUserQueryHandler) {
        this.getUserQueryHandler = getUserQueryHandler;
    }

    public User dispatch(GetUserQuery query) {
        return getUserQueryHandler.handle(query);
    }
}Code language: Java (java)

5. Integrating with Event Sourcing

What is Event Sourcing?

Event Sourcing is a pattern where state changes are stored as a sequence of events. Instead of storing the current state, you store the events that lead to the current state. This provides a complete audit log and can be useful for reconstructing past states.

Implementing Event Sourcing

To integrate event sourcing, we need to define events and an event store.

Event Model

Create an event for user creation:

package com.example.event;

public class UserCreatedEvent {
    private final String userId;
    private final String username;
    private final String email;

    public UserCreatedEvent(String userId, String username, String email) {
        this.userId = userId;
        this.username = username;
        this.email = email;
    }

    public String getUserId() {
        return userId;
    }

    public String getUsername() {
        return username;
    }

    public String getEmail() {
        return email;
    }
}Code language: Java (java)

Event Store

The event store saves events and can replay them to reconstruct the state.

package com.example.event;

import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

@Service
public class EventStore {

    private final List<UserCreatedEvent> events = new ArrayList<>();

    public void saveEvent(UserCreatedEvent event) {
        events.add(event);
    }

    public List<UserCreatedEvent> getEvents() {
        return events;
    }
}Code language: Java (java)

Modifying Command Handler

Modify the CreateUserCommandHandler to save events:

package com.example.command;

import com.example.event.EventStore

;
import com.example.event.UserCreatedEvent;
import com.example.model.User;
import com.example.service.UserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class CreateUserCommandHandler {

    private final UserService userService;
    private final EventStore eventStore;

    @Autowired
    public CreateUserCommandHandler(UserService userService, EventStore eventStore) {
        this.userService = userService;
        this.eventStore = eventStore;
    }

    public void handle(CreateUserCommand command) {
        User user = new User(command.getUserId(), command.getUsername(), command.getEmail());
        userService.save(user);

        UserCreatedEvent event = new UserCreatedEvent(command.getUserId(), command.getUsername(), command.getEmail());
        eventStore.saveEvent(event);
    }
}Code language: Java (java)

6. Handling Concurrency

Concurrency can be a challenge in CQRS systems, especially when multiple commands are trying to modify the same data. There are two main strategies for handling concurrency: optimistic and pessimistic concurrency control.

Optimistic Concurrency Control

In optimistic concurrency control, you assume that conflicts are rare and only check for conflicts before committing the transaction.

Add a version field to the user entity:

package com.example.model;

import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Version;

@Entity
public class User {

    @Id
    private String userId;
    private String username;
    private String email;

    @Version
    private Long version;

    // Constructors, getters, and setters omitted for brevity
}Code language: Java (java)

Modify the user service to handle version conflicts:

package com.example.service;

import com.example.model.User;
import com.example.repository.UserRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.stereotype.Service;

import java.util.Optional;

@Service
public class UserService {

    private final UserRepository userRepository;

    @Autowired
    public UserService(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    public void save(User user) {
        try {
            userRepository.save(user);
        } catch (OptimisticLockingFailureException e) {
            // Handle version conflict
        }
    }

    public User findById(String userId) {
        Optional<User> user = userRepository.findById(userId);
        return user.orElse(null);
    }
}Code language: Java (java)

Pessimistic Concurrency Control

In pessimistic concurrency control, you lock the data when a transaction starts and release the lock when the transaction ends.

Modify the user service to use pessimistic locking:

package com.example.service;

import com.example.model.User;
import com.example.repository.UserRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.persistence.EntityManager;
import javax.persistence.LockModeType;

@Service
public class UserService {

    private final UserRepository userRepository;
    private final EntityManager entityManager;

    @Autowired
    public UserService(UserRepository userRepository, EntityManager entityManager) {
        this.userRepository = userRepository;
        this.entityManager = entityManager;
    }

    public void save(User user) {
        entityManager.lock(user, LockModeType.PESSIMISTIC_WRITE);
        userRepository.save(user);
    }

    public User findById(String userId) {
        return entityManager.find(User.class, userId, LockModeType.PESSIMISTIC_READ);
    }
}Code language: Java (java)

7. Putting It All Together

Example Use Case

Let’s put everything together with an example use case of creating and querying a user.

Create a User Controller

package com.example.controller;

import com.example.bus.CommandBus;
import com.example.command.CreateUserCommand;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/users")
public class UserController {

    private final CommandBus commandBus;

    @Autowired
    public UserController(CommandBus commandBus) {
        this.commandBus = commandBus;
    }

    @PostMapping
    public void createUser(@RequestBody CreateUserCommand command) {
        commandBus.dispatch(command);
    }
}Code language: Java (java)

Query a User Controller

package com.example.controller;

import com.example.bus.QueryBus;
import com.example.model.User;
import com.example.query.GetUserQuery;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/users")
public class UserQueryController {

    private final QueryBus queryBus;

    @Autowired
    public UserQueryController(QueryBus queryBus) {
        this.queryBus = queryBus;
    }

    @GetMapping("/{userId}")
    public User getUser(@PathVariable String userId) {
        return queryBus.dispatch(new GetUserQuery(userId));
    }
}Code language: Java (java)

Testing the Implementation

Write integration tests to verify the implementation.

Test Creating a User

package com.example;

import com.example.command.CreateUserCommand;
import com.example.event.EventStore;
import com.example.model.User;
import com.example.repository.UserRepository;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;

@SpringBootTest
public class CreateUserCommandTest {

    @Autowired
    private UserRepository userRepository;

    @Autowired
    private EventStore eventStore;

    @Test
    public void testCreateUser() {
        CreateUserCommand command = new CreateUserCommand("1", "john_doe", "[email protected]");
        commandBus.dispatch(command);

        User user = userRepository.findById("1").orElse(null);
        assertNotNull(user);
        assertEquals("john_doe", user.getUsername());
        assertEquals("[email protected]", user.getEmail());

        UserCreatedEvent event = eventStore.getEvents().get(0);
        assertEquals("1", event.getUserId());
        assertEquals("john_doe", event.getUsername());
        assertEquals("[email protected]", event.getEmail());
    }
}Code language: Java (java)

Test Querying a User

package com.example;

import com.example.model.User;
import com.example.repository.UserRepository;
import com.example.query.GetUserQuery;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;

@SpringBootTest
public class GetUserQueryTest {

    @Autowired
    private UserRepository userRepository;

    @Autowired
    private QueryBus queryBus;

    @Test
    public void testGetUser() {
        User user = new User("1", "john_doe", "[email protected]");
        userRepository.save(user);

        User result = queryBus.dispatch(new GetUserQuery("1"));
        assertNotNull(result);
        assertEquals("john_doe", result.getUsername());
        assertEquals("[email protected]", result.getEmail());
    }
}Code language: Java (java)

8. Conclusion

Implementing CQRS in Java involves creating separate models and handlers for commands and queries, integrating with event sourcing for better state management, and handling concurrency to ensure data consistency. This tutorial provided a step-by-step guide to implementing CQRS in a Java application, highlighting the benefits and challenges of this design pattern.

By following these steps, you can build scalable, maintainable, and performant applications that leverage the power of CQRS. Remember that while CQRS offers many advantages, it also introduces complexity, so it’s essential to evaluate whether it fits your specific use case before adopting it.