CQRS框架：AxonFramework 之 Hello World - yjmyzz－IT工程師數位筆記本

文章出處

Command Query Responsibility Segregation,CQRS 這個架構好象最近博客園里討論得比較多，有幾篇園友的文章很有深度，推薦閱讀：

CQRS架構簡介

淺談命令查詢職責分離(CQRS)模式

DDD CQRS架構和傳統架構的優缺點比較

比較有趣的是，以往一斷談及架構思路、OO這些，往往都是java大佬們的專長，而CQRS這個話題，好象.NET占了上風。園友湯雪華的ENODE開源大作，在github上人氣也很旺。

于是，我逆向思路搜索了下java的類似項目，果然有一個AxonFramework，甚至還有一個專門的網站。按文檔上的介紹，弄了一個hello world，記錄一下：

CRQS是基于事件驅動的，其主要架構并不復雜，見下圖：

點擊看大圖

簡單來講，對數據庫的修改操作，UI層只管發送各種命令(Command)，觸發事件(Event)，然后由EventHandler去異步處理，最終寫入master DB，對于數據庫的查詢，則查詢slave DB（注：這里的master db, slave db只是一個邏輯上的區分，可以是真正的主-從庫，也可以都是一個庫）。這樣的架構，很容易實現讀寫分離，也易于大型項目的擴展。

項目結構：

點擊看大圖

package的名稱上大概就能看出用途：

command包定義各種命令，

event包定義各種事件，

handler包定義事件處理邏輯，

model包相當于領域模型

最外層的ToDOItemRunner相當于應用程序入口。

gradle依賴項：

group 'yjmyzz'
version '1.0'

apply plugin: 'java'
apply plugin: 'application'

sourceCompatibility = 1.8

repositories {
    mavenLocal()
    maven {
        url 'http://maven.oschina.net/content/groups/public/'
    }
    mavenCentral()
}

dependencies {
    testCompile group: 'junit', name: 'junit', version: '4.12'
    compile "org.axonframework:axon:2.4.3"
    compile "org.axonframework:axon-core:2.4.3"
    compile "org.axonframework:axon-test:2.4.3"
    compile 'org.springframework:spring-core:4.2.3.RELEASE'
    compile 'org.springframework:spring-beans:4.2.3.RELEASE'
    compile 'org.springframework:spring-context:4.2.3.RELEASE'
    compile 'org.springframework:spring-context-support:4.2.3.RELEASE'
    compile 'org.springframework:spring-aop:4.2.3.RELEASE'
    compile 'org.springframework:spring-test:4.2.3.RELEASE'
    compile 'org.apache.logging.log4j:log4j-slf4j-impl:2.5'
    compile 'org.apache.logging.log4j:log4j-core:2.5'
    compile 'javax.persistence:persistence-api:2.1'
}

mainClassName='demo.axon.ToDoItemRunner'

command命令：

這里我們假設了二個命令：創建命令、完成命令

CreateToDoItemCommand

package demo.axon.command;

import org.axonframework.commandhandling.annotation.TargetAggregateIdentifier;

public class CreateToDoItemCommand {

    @TargetAggregateIdentifier
    private final String todoId;
    private final String description;

    public CreateToDoItemCommand(String todoId, String description) {
        this.todoId = todoId;
        this.description = description;
    }

    public String getTodoId() {
        return todoId;
    }

    public String getDescription() {
        return description;
    }
}

MarkCompletedCommand

package demo.axon.command;

import org.axonframework.commandhandling.annotation.TargetAggregateIdentifier;

public class MarkCompletedCommand {

    @TargetAggregateIdentifier
    private final String todoId;

    public MarkCompletedCommand(String todoId) {
        this.todoId = todoId;
    }

    public String getTodoId() {
        return todoId;
    }
}

Event事件：

ToDoItemCreatedEvent

package demo.axon.event;

public class ToDoItemCreatedEvent {

    private final String todoId;
    private final String description;

    public ToDoItemCreatedEvent(String todoId, String description) {
        this.todoId = todoId;
        this.description = description;
    }

    public String getTodoId() {
        return todoId;
    }

    public String getDescription() {
        return description;
    }
}

ToDoItemCompletedEvent

package demo.axon.event;

public class ToDoItemCompletedEvent {

    private final String todoId;

    public ToDoItemCompletedEvent(String todoId) {
        this.todoId = todoId;
    }

    public String getTodoId() {
        return todoId;
    }
}

EventHandler事件處理

package demo.axon.handler;

import demo.axon.event.ToDoItemCompletedEvent;
import demo.axon.event.ToDoItemCreatedEvent;
import org.axonframework.eventhandling.annotation.EventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ToDoEventHandler {

    Logger logger = LoggerFactory.getLogger(ToDoEventHandler.class);

    @EventHandler
    public void handle(ToDoItemCreatedEvent event) {
        logger.info("We've got something to do: " + event.getDescription() + " (" + event.getTodoId() + ")");
    }

    @EventHandler
    public void handle(ToDoItemCompletedEvent event) {
        logger.info("We've completed a task: " + event.getTodoId());
    }
}

上面的代碼只是演示，將事件信息輸出而已，真實應用中，這里可以完成對db的更新操作。　

領域模型model

package demo.axon.model;

import demo.axon.command.CreateToDoItemCommand;
import demo.axon.command.MarkCompletedCommand;
import demo.axon.event.ToDoItemCompletedEvent;
import demo.axon.event.ToDoItemCreatedEvent;
import org.axonframework.commandhandling.annotation.CommandHandler;
import org.axonframework.eventhandling.annotation.EventHandler;
import org.axonframework.eventsourcing.annotation.AbstractAnnotatedAggregateRoot;
import org.axonframework.eventsourcing.annotation.AggregateIdentifier;

public class ToDoItem extends AbstractAnnotatedAggregateRoot {

    @AggregateIdentifier
    private String id;

    public ToDoItem() {
    }

    @CommandHandler
    public ToDoItem(CreateToDoItemCommand command) {
        apply(new ToDoItemCreatedEvent(command.getTodoId(), command.getDescription()));
    }

    @EventHandler
    public void on(ToDoItemCreatedEvent event) {
        this.id = event.getTodoId();
    }

    @CommandHandler
    public void markCompleted(MarkCompletedCommand command) {
        apply(new ToDoItemCompletedEvent(id));
    }
}

然后讓Spring將這些東西串在一起，配置文件如下：

 1 <?xml version="1.0" encoding="UTF-8"?>
 2 <beans xmlns="http://www.springframework.org/schema/beans"
 3        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 4        xmlns:axon="http://www.axonframework.org/schema/core"
 5        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
 6                            http://www.axonframework.org/schema/core http://www.axonframework.org/schema/axon-core-2.0.xsd">
 7 
 8     <axon:command-bus id="commandBus"/>
 9     <axon:event-bus id="eventBus"/>
10 
11     <axon:event-sourcing-repository id="toDoRepository"
12                                     aggregate-type="demo.axon.model.ToDoItem"/>
13 
14     <axon:aggregate-command-handler id="toDoItemHandler"
15                                     aggregate-type="demo.axon.model.ToDoItem"
16                                     repository="toDoRepository"
17                                     command-bus="commandBus"/>
18 
19     <axon:filesystem-event-store id="eventStore" base-dir="events"/>
20 
21     <axon:annotation-config />
22 
23     <bean class="demo.axon.handler.ToDoEventHandler"/>
24 
25     <bean class="org.axonframework.commandhandling.gateway.CommandGatewayFactoryBean">
26         <property name="commandBus" ref="commandBus"/>
27     </bean>
28 
29 </beans>

View Code

最后，提供一個舞臺，讓整個應用run起來：

package demo.axon;

import demo.axon.command.CreateToDoItemCommand;
import demo.axon.command.MarkCompletedCommand;
import org.axonframework.commandhandling.gateway.CommandGateway;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.util.UUID;

public class ToDoItemRunner {

    private CommandGateway commandGateway;

    public ToDoItemRunner(CommandGateway commandGateway) {
        this.commandGateway = commandGateway;
    }

    public static void main(String[] args) {
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("sampleContext.xml");
        ToDoItemRunner runner = new ToDoItemRunner(applicationContext.getBean(CommandGateway.class));
        runner.run();
    }

    private void run() {
        final String itemId = UUID.randomUUID().toString();
        commandGateway.send(new CreateToDoItemCommand(itemId, "Need to do this"));
        commandGateway.send(new MarkCompletedCommand(itemId));
    }
}

輸出結果：

12:01:36,113 <demo.axon.handler.ToDoEventHandler>  INFO [main]: We've got something to do: Need to do this (3126f293-67fd-4bb7-b152-069acba775b6)
12:01:36,114 <org.axonframework.commandhandling.callbacks.LoggingCallback>  INFO [main]: Command executed successfully: demo.axon.command.CreateToDoItemCommand
12:01:36,205 <demo.axon.handler.ToDoEventHandler>  INFO [main]: We've completed a task: 3126f293-67fd-4bb7-b152-069acba775b6
12:01:36,206 <org.axonframework.commandhandling.callbacks.LoggingCallback>  INFO [main]: Command executed successfully: demo.axon.command.MarkCompletedCommand

axon框架測試也很容易：

package test.demo.axon;

import demo.axon.command.CreateToDoItemCommand;
import demo.axon.command.MarkCompletedCommand;
import demo.axon.event.ToDoItemCompletedEvent;
import demo.axon.event.ToDoItemCreatedEvent;
import demo.axon.model.ToDoItem;
import org.axonframework.test.FixtureConfiguration;
import org.axonframework.test.Fixtures;
import org.junit.Before;
import org.junit.Test;

public class ToDoItemTest {

    private FixtureConfiguration fixture;

    @Before
    public void setUp() throws Exception {
        fixture = Fixtures.newGivenWhenThenFixture(ToDoItem.class);
    }

    @Test
    public void testCreateToDoItem() throws Exception {
        fixture.given()
                .when(new CreateToDoItemCommand("todo1", "need to implement the aggregate"))
                .expectEvents(new ToDoItemCreatedEvent("todo1", "need to implement the aggregate"));
    }

    @Test
    public void testMarkToDoItemAsCompleted() throws Exception {
        fixture.given(new ToDoItemCreatedEvent("todo1", "need to implement the aggregate"))
                .when(new MarkCompletedCommand("todo1"))
                .expectEvents(new ToDoItemCompletedEvent("todo1"));
    }

}

given/when/expectEvents的意思是，給(given)一個事件，然后當(when)某個命令被調用時，期待(expectEvents)某個事件被觸發。

最后 github上還有一個比較復雜的示例項目：https://github.com/AxonFramework/Axon-trader，想深入了解的可以研究下

文章列表