原文はこちら。
https://community.oracle.com/community/cloud_computing/oracle-cloud-developer-solutions/blog/2017/03/30/develop-microservices-application-using-cqrs-and-event-sourcing-patterns
このエントリでは、CQRSとイベントソーシングパターンを使用して、イベント駆動のマイクロサービスアプリケーションを構築する方法を紹介します。 以下は、このエントリで説明する概念を簡潔にまとめたものです。詳細は、このブログの最後にあるリソースから入手してください。
What is a Microservice?
このアーキテクチャースタイルにはずばりこれ、という定義はありませんが、Adrian Cockcroftによれば、マイクロサービスアーキテクチャは、境界付けられたコンテキストを持つ要素を疎結合で構成したサービス指向アーキテクチャと定義しています。
Adrian Cockcroftについて
http://www.battery.com/our-team/member/adrian-cockcroft/
What is a Bounded Context?
Bounded Context（境界付けられたコンテキスト）とは、ドメインモデルやデータモデル、アプリケーションサービスといったシングルドメインの詳細をカプセル化するというコンセプトで、別の境界付けられたコンテキストやドメインとの統合ポイントを定義します。
What is CQRS?
Command Query Responsibility Segregation (CQRS) とは、ドメイン操作をクエリとコマンドの2つのカテゴリに分けるアーキテクチャパターンです。クエリは状態を変更せずに結果を返すだけですが、コマンドはドメインモデルの状態を変更する操作です。
Why CQRS?
アプリケーション・ライフサイクルでは、論理モデルがより複雑で構造化されると、ユーザーエクスペリエンスに影響を与える可能性があるため、ユーザーエクスペリエンスは、コアシステムとは独立していなければなりません。
スケーラブルでアプリケーションのメンテナンスを簡単にするため、読み取りモデルと書き込みモデル間の制約を減らす必要があります。読み書きを分ける理由は以下の通りです。
- スケーラビリティ
読み取りが書き込みを上回るので、それぞれのスケール要件が異なり、より適切に対応できる
- フレキシビリティ
別々の読み取り/書き込みモデル
- 複雑性の軽減
複雑さを別々の問題にシフト
What is Event sourcing?
イベントソーシングは、異なるイベント中心のアプローチを使ってビジネスエンティティを永続化することで、アトミック性を実現します。
WhyEventSourcing
https://github.com/cer/event-sourcing-examples/wiki/WhyEventSourcing
エンティティの現在の状態を格納するのではなく、アプリケーションは、エンティティの状態を変更した一連の「イベント」を格納します。アプリケーションは、イベントを再生してエンティティの現在の状態を再構築できます。イベントの保存は単一の操作であるため、本質的にアトミックであり、分散トランザクションに通常関連付けられる2PC（2フェーズコミット）を必要としません。
Overview
このエントリでは、CQRSとイベントソーシングパターンを適用して、追加、削除、および読み取り操作を伴う「カート」という単一の境界付けられたコンテキストで構成される単純なマイクロサービスアプリケーションを開発する方法について説明します。このサンプルには機能的な意味はありませんが、基礎となるパターンとその実装を理解するのに十分でしょう。次の図は、CQRSおよびイベントソーシングパターンを使用してアプリケーションを構築する際のアクティビティの概念フローを示しています。
 |
| Figure 1 CQRS and Event sourcing |
このエントリで紹介しているサンプルで利用しているテクノロジースタックは以下の通りです。
このような前提を踏まえて、サンプルの構築をすすめていきましょう。
Identify Aggregate Root
第1のステップは、境界付けられたコンテキストを識別し、境界付けられたコンテキストのドメインエンティティを識別することです。これは（'account'、 'order' ...といった）Aggregate Root（集約ルート）を定義するのに役立ちます。Aggregate（集約）とは、常に一貫性のある状態に保たれるエンティティまたはエンティティのグループです。Aggregate Rootは、この一貫性のある状態を維持する責任を負う集約ツリーの最上位にあるオブジェクトです。
今回は簡単のため、ドメインモデルの唯一のAggregate Rootとして「カート」を考えます。 通常のショッピングカートのように、カート内のアイテムは、そのカートへの追加または削除されたものによって調整されます。
Define Commands
このAggregate Rootには2個のコマンドがあります。
- Cartへの追加Command
AddToCartCommand クラスでモデリング
- Cartからの削除Command
RemoveFromCartCommand クラスでモデリング
public class AddToCartCommand {
private final String cartId;
private final int item;
public AddToCartCommand(String cartId, int item) {
this.cartId = cartId;
this.item = item;
}
public String getCartId() {
return cartId;
}
public int getItem() {
return item;
}
}
public class RemoveFromCartCommand {
private final String cartId;
private final int item;
public RemoveFromCartCommand(String cartId, int item) {
this.cartId = cartId;
this.item = item;
}
public String getCartId() {
return cartId;
}
public int getItem() {
return item;
}
}
|
ご存じのとおり、これらのコマンドはただのPOJOで、システム内で何が起こる必要があるのか、また必要な情報とともにシステム内で発生する必要のあるものごとを捕捉するために使います。Axonフレームワークでは、インターフェイスの実装やクラスの拡張を行うコマンドは必要ありません。
Define Command Handlers
コマンドにはハンドラが1つだけあります。以下のクラスはカートへの追加およびカートからの削除コマンドのハンドラを表します。
@Component
public class AddToCartCommandHandler {
private Repository repository;
@Autowired
public AddToCartCommandHandler(Repository repository) {
this.repository = repository;
}
@CommandHandler
public void handle(AddToCartCommand addToCartCommand){
Cart cartToBeAdded = (Cart) repository.load(addToCartCommand.getCartId());
cartToBeAdded.addCart(addToCartCommand.getItem());
}
}
@Component
public class RemoveFromCartHandler {
private Repository repository;
@Autowired
public RemoveFromCartHandler(Repository repository) {
this.repository = repository;
}
@CommandHandler
public void handle(RemoveFromCartCommand removeFromCartCommand){
Cart cartToBeRemoved = (Cart) repository.load(removeFromCartCommand.getCartId());
cartToBeRemoved.removeCart(removeFromCartCommand.getItem());
}
}
|
Spring BootともにAxonを使用しているため、上で定義したSpring Beanには@CommandHandlerというアノテーションが付いたメソッドがあり、これをコマンドハンドラとして扱うことができます。@Componentアノテーションは、アプリケーション起動時にこれらのBeanをスキャンし、任意のAuto wired ResourceをこのBeanに注入します。Aggregateに直接アクセスするのではなく、AxonフレームワークのドメインオブジェクトであるRepositoryが、Aggregateの取得と永続化を抽象化します。
Application Startup
以下のAppConfigurationクラスは、アプリケーション・デプロイ時に初期化され、パターン実装時に必要なコンポーネントを作成するSpring構成クラスです。
@Configuration
@AnnotationDriven
public class AppConfiguration {
@Bean
public DataSource dataSource() {
return DataSourceBuilder
.create()
.username("sa")
.password("")
.url("jdbc:h2:mem:axonappdb")
.driverClassName("org.h2.Driver")
.build();
}
/**
* Event store to store events
*/
@Bean
public EventStore jdbcEventStore() {
return new JdbcEventStore(dataSource());
}
@Bean
public SimpleCommandBus commandBus() {
SimpleCommandBus simpleCommandBus = new SimpleCommandBus();
return simpleCommandBus;
}
/**
* Cluster event handlers that listens to events thrown in the application.
*/
@Bean
public Cluster normalCluster() {
SimpleCluster simpleCluster = new SimpleCluster("simpleCluster");
return simpleCluster;
}
/**
* This configuration registers event handlers with defined clusters
*/
@Bean
public ClusterSelector clusterSelector() {
Map<String, Cluster> clusterMap = new HashMap<>();
clusterMap.put("msacqrses.eventhandler", normalCluster());
return new ClassNamePrefixClusterSelector(clusterMap);
}
/**
*The clustering event bus is needed to route events to event handlers in the clusters.
*/
@Bean
public EventBus clusteringEventBus() {
ClusteringEventBus clusteringEventBus = new ClusteringEventBus(clusterSelector(), terminal());
return clusteringEventBus;
}
/**
* Event Bus Terminal publishes domain events to the cluster
*
*/
@Bean
public EventBusTerminal terminal() {
return new EventBusTerminal() {
@Override
public void publish(EventMessage... events) {
normalCluster().publish(events);
}
@Override
public void onClusterCreated(Cluster cluster) {
}
};
}
/**
* Command gateway through which all commands in the application are submitted
*
*/
@Bean
public DefaultCommandGateway commandGateway() {
return new DefaultCommandGateway(commandBus());
}
/**
* Event Repository that handles retrieving of entity from the stream of events.
*/
@Bean
public Repository<Cart> eventSourcingRepository() {
EventSourcingRepository eventSourcingRepository = new EventSourcingRepository(Cart.class, jdbcEventStore());
eventSourcingRepository.setEventBus(clusteringEventBus());
return eventSourcingRepository;
}
}
|
Axonが提供する、このクラスで初期化された主要なインフラストラクチャコンポーネントを見てみましょう。
Command bus
図1の通り、command bus（コマンドバス）は、それぞれのコマンドハンドラにコマンドをルーティングするコンポーネントです。Axon Frameworkには、コマンドをコマンド・ハンドラに渡すために使用可能な種々のCommand Busが付属しています。 AxonのCommand Busの実装の詳細は以下のURLを参照してください。
Axon Framework 2.0.9 Reference Guide
Command Handling
http://www.axonframework.org/docs/2.0/command-handling.html
この例では、SpringのアプリケーションコンテキストでBeanとして構成されているSimpleCommandBusを使用します。
Command Gateway
Command Bus直接コマンドを送信できますが、通常はCommand Gatewayを使用することをお勧めします。Command Gatewayを使用することで、開発者はコマンドのインターセプト、障害時の再試行の設定といった機能を実行できます。この例では、Command Busを直接使用せず、コマンドを送信するSpring Beanとして構成されているDefaultCommandGatewayを使用します。これはAxonがデフォルトで提供しているものです。
Event Bus
図1の通り、Aggregate Rootで開始されたコマンドは、イベントとして送信され、Event Storeで永続化されます。Event Busは、イベントをイベントハンドラにルーティングする基盤です。Event Busはメッセージディスパッチの観点からCommand Busに似ていますが、基本的に異なるものです。
Command Busは、近い将来に何が起こるかを定義するコマンドとともに動作し、コマンドを解釈するコマンドハンドラは1つだけです。それに対し、Event Busでは、イベントをルーティングし、イベントに対してゼロ個以上のハンドラを使って、過去に発生したアクションを定義します。
Axonは、Event Busの複数の実装を定義していますが、今回はSpring Beanとして再度ワイヤリングされたClusteringEventBusを使用します。AxonのEvent Bus実装の詳細については、以下のURLを参照してください。
Axon Framework 2.0.9 Reference Guide
6. Event Processing
http://www.axonframework.org/docs/2.0/event-processing.html
Event Store
リポジトリはドメインオブジェクトの現在の状態ではなくドメインイベントを格納するため、イベントストアを構成する必要があります。Axon Frameworkでは、JDBC、JPA、ファイルシステムなどの複数の永続化方式を使用してイベントを格納できます。今回は、JDBCイベントストアを使用します。
Event Sourcing Repository
今回はAggregate Rootを永続メカニズムの表現から作成せず、Event sourcingリポジトリを使って構成できるイベントストリームから作成します。ドメインイベントを発行する予定であるため、前に定義したEvent Busを使用してリポジトリを設定します。
Database
今回はデータストアとしてインメモリデータベース（h2）を使います。Spring Bootのapplication.propertiesにはデータソース構成の設定が含まれています。
spring.datasource.url=jdbc:h2:mem:axonappdb
spring.datasource.driverClassName=org.h2.Driver
spring.datasource.username=sa
spring.datasource.password=
spring.datasource.validation-query=SELECT 1;
spring.datasource.initial-size=2
spring.datasource.sql-script-encoding=UTF-8
spring.jpa.database=h2
spring.jpa.show-sql=true
spring.jpa.hibernate.ddl-auto=create
|
前述の通り、この例ではJDBCイベントストアを使ってシステムで生成されるイベントを格納します。これらのイベントをAxon Frameworkが指定する（Axon Frameworkイベントインフラストラクチャの）デフォルト表に格納します。以下のスタートアップクラスを使って、この例で必要なデータベース表を作成します。
@Component
public class Datastore {
@Autowired
@Qualifier("transactionManager")
protected PlatformTransactionManager txManager;
@Autowired
private Repository repository;
@Autowired
private javax.sql.DataSource dataSource;
@PostConstruct
private void init() {
TransactionTemplate transactionTmp = new TransactionTemplate(txManager);
transactionTmp.execute(new TransactionCallbackWithoutResult() {
@Override
protected void doInTransactionWithoutResult(TransactionStatus status) {
UnitOfWork uow = DefaultUnitOfWork.startAndGet();
repository.add(new Cart("cart1"));
repository.add(new Cart("cart2"));
uow.commit();
}
});
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
jdbcTemplate.execute("create table cartview (cartid VARCHAR , items NUMBER )");
jdbcTemplate.update("insert into cartview (cartid, items) values (?, ?)", new Object[]{"cart1", 0});
jdbcTemplate.update("insert into cartview (cartid, items) values (?, ?)", new Object[]{"cart2", 0});
}
...
}
|
このスタートアップクラスは、コマンド処理に使う2個のカートエントリをリポジトリに作成し、クエリ処理に使うデータベース表（cartview）を作成します。
ここまでで実施したことをまとめておきます。
- Aggregate RootとしてCartを識別し、Cartへのアイテム追加および削除のコマンドおよびコマンドハンドラを定義した
- CQRS およびイベントソーシングに必要な基盤コンポーネントを初期化するスタートアップクラスを定義した
- データベース表の作成ならびにこのサンプルが必要とするデータ設定のためのスタートアップクラスも定義した
以下で定義するAggregateRootの"Cart"を見ていきましょう。
Aggregate Root
public class Cart extends AbstractAnnotatedAggregateRoot {
@AggregateIdentifier
private String cartid;
private int items;
public Cart() {
}
public Cart(String cartId) {
apply(new CartCreatedEvent(cartId));
}
@EventSourcingHandler
public void applyCartCreation(CartCreatedEvent event) {
this.cartid = event.getCartId();
this.items = 0;
}
public void removeCart(int removeitem) {
/**
* State is not directly changed, we instead apply event that specifies what happened. Events applied are stored.
*/
if(this.items > removeitem && removeitem > 0)
apply(new RemoveFromCartEvent(this.cartid, removeitem, this.items));
}
@EventSourcingHandler
private void applyCartRemove(RemoveFromCartEvent event) {
/**
* When events stored in the event store are applied on an Entity this method is
* called. Once all the events in the event store are applied, it will bring the Cart * to the most recent state.
*/
this.items -= event.getItemsRemoved();
}
public void addCart(int item) {
/**
* State is not directly changed, we instead apply event that specifies what happened. Events applied are stored.
*/
if(item > 0)
apply(new AddToCartEvent(this.cartid, item, this.items));
}
@EventSourcingHandler
private void applyCartAdd(AddToCartEvent event) {
/**
* When events stored in the event store are applied on an Entity this method is
* called. Once all the events in the event store are applied, it will bring the
* Cart to the most recent state.
*/
this.items += event.getItemAdded();
}
public int getItems() {
return items;
}
public void setIdentifier(String id) {
this.cartid = id;
}
@Override
public Object getIdentifier() {
return cartid;
}
}
|
以下はAggregate Rootの定義のキーポイントです。
- @AggregateIdentifier は、エンティティのIDを示すフィールドをマークする、JPAにおける @Id と似ている
- ドメイン駆動デザインでは、ドメインエンティティに関連するビジネスロジックを含めることを推奨しており、そのロジックが上記定義にあるビジネスメソッドである。詳細はReferencesのセクションを参照のこと。
- コマンドが呼び出されると、ドメインオブジェクトをリポジトリから取得し、それぞれのメソッド（例えばaddCart）がそのドメインオブジェクト（ここではCart）で呼び出される。
- ドメインオブジェクトは直接状態を変更せずに、適切なイベントを適用する
- イベントはイベントストアに保存され、そのイベントに対応するハンドラが呼び出され、ドメインオブジェクトの変更が発生する。
- CartというAggregate Rootは更新（つまりコマンドによる状態変更）にのみ利用されることに注意する必要がある。全てのクエリリクエストは、（次セクションで説明する）別のデータベースエンティティで処理される。
ではイベントならびにCartエンティティから呼び出されるドメインイベントを管理するイベントハンドラを見ていきましょう。
Events
前のセクションで説明したように、Cartエンティティで呼び出される2つのコマンド、Cartへの追加とCartからの削除があります。これらのコマンドがAggregate Rootで実行されると、以下に示すAddToCartEventとRemoveFromCartEventという2つのイベントを生成します。
public class AddToCartEvent {
private final String cartId;
private final int itemAdded;
private final int items;
private final long timeStamp;
public AddToCartEvent(String cartId, int itemAdded, int items) {
this.cartId = cartId;
this.itemAdded = itemAdded;
this.items = items;
ZoneId zoneId = ZoneId.systemDefault();
this.timeStamp = LocalDateTime.now().atZone(zoneId).toEpochSecond();
}
public String getCartId() {
return cartId;
}
public int getItemAdded() {
return itemAdded;
}
public int getItems() {
return items;
}
public long getTimeStamp() {
return timeStamp;
}
}
public class RemoveFromCartEvent {
private final String cartId;
private final int itemsRemoved;
private final int items;
private final long timeStamp;
public RemoveFromCartEvent(String cartId, int itemsRemoved, int items) {
this.cartId = cartId;
this.itemsRemoved = itemsRemoved;
this.items = items;
ZoneId zoneId = ZoneId.systemDefault();
this.timeStamp = LocalDateTime.now().atZone(zoneId).toEpochSecond();
}
public String getCartId() {
return cartId;
}
public int getItemsRemoved() {
return itemsRemoved;
}
public int getItems() {
return items;
}
public long getTimeStamp() {
return timeStamp;
}
}
|
Event Handlers
上述のイベントは以下のイベントハンドラで処理されます。
@Component
public class AddToCartEventHandler {
@Autowired
DataSource dataSource;
@EventHandler
public void handleAddToCartEvent(AddToCartEvent event, Message msg) {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
String cartId = event.getCartId();
int items = event.getItems();
int itemToBeAdded = event.getItemAdded();
int newItems = items + itemToBeAdded;
String updateQuery = "UPDATE cartview SET items = ? WHERE cartid = ?";
jdbcTemplate.update(updateQuery, new Object[]{newItems, cartId});
}
}
@Component
public class RemoveFromCartEventHandler {
@Autowired
DataSource dataSource;
@EventHandler
public void handleRemoveFromCartEvent(RemoveFromCartEvent event) {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
String cartId = event.getCartId();
int items = event.getItems();
int itemsToBeRemoved = event.getItemsRemoved();
int newItems = items - itemsToBeRemoved;
String update = "UPDATE cartview SET items = ? WHERE cartid = ?";
jdbcTemplate.update(update, new Object[]{newItems, cartId});
}
}
|
お気づきの通り、イベントハンドラはcartviewというデータベース表を更新します。この表はCartエンティティをクエリする際に使われます。コマンドをあるドメインで実行している間、イベントソーシングでCQRSを実現することにより、クエリリクエストは別のドメインで処理されます。
Controllers
この例では、2個のSpringコントローラクラスを定義しています。一つはCartドメインの更新、一つはCartドメインのクエリのためのクラスです。以下のようにブラウザからRESTエンドポイントを呼び出すことができます。
- http://<host>:<port>/add/cart/<noOfItems>
- http://<host>:<port>/remove/cart/<noOfItems>
- http://<host>:<port>/view
@RestController
public class CommandController {
@Autowired
private CommandGateway commandGateway;
@RequestMapping("/remove/{cartId}/{item}")
@Transactional
public ResponseEntity doRemove(@PathVariable String cartId, @PathVariable int item) {
RemoveFromCartCommand removeCartCommand = new RemoveFromCartCommand(cartId, item);
commandGateway.send(removeCartCommand);
return new ResponseEntity<>("Remove event generated. Status: "+ HttpStatus.OK, HttpStatus.OK);
}
@RequestMapping("/add/{cartId}/{item}")
@Transactional
public ResponseEntity doAdd(@PathVariable String cartId, @PathVariable int item) {
AddToCartCommand addCartCommand = new AddToCartCommand(cartId, item);
commandGateway.send(addCartCommand);
return new ResponseEntity<>("Add event generated. Status: "+ HttpStatus.OK, HttpStatus.OK);
}
}
@RestController
public class ViewController {
@Autowired
private DataSource dataSource;
@RequestMapping(value = "/view", method = RequestMethod.GET, produces = MediaType.APPLICATION_JSON_VALUE)
public ResponseEntity getItems() {
JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
List<Map<String, Integer>> queryResult = jdbcTemplate.query("SELECT * from cartview ORDER BY cartid", (rs, rowNum) -> {
return new HashMap<String, Integer>() {{
put(rs.getString("CARTID"), rs.getInt("ITEMS"));
}};
});
if (queryResult.size() > 0) {
return new ResponseEntity<>(queryResult, HttpStatus.OK);
}
else {
return new ResponseEntity<>(null, HttpStatus.NOT_FOUND);
}
}
}
|
Deployment
Spring Bootを使ってアプリケーションを実行可能なJarファイルとしてパッケージングし、アプリケーションをOracle Application Container Cloud Serviceにデプロイします。以下のSpring Bootクラスでアプリケーションを初期化します。
@SpringBootApplication
public class AxonApp {
public static final Optional<String> host;
public static final Optional<String> port;
public static final Properties myProps = new Properties();
static {
host = Optional.ofNullable(System.getenv("HOSTNAME"));
port = Optional.ofNullable(System.getenv("PORT"));
}
public static void main(String[] args) {
myProps.setProperty("server.address", host.orElse("localhost"));
myProps.setProperty("server.port", port.orElse("8128"));
SpringApplication app = new SpringApplication(AxonApp.class);
app.setDefaultProperties(myProps);
app.run(args);
}
}
|
以下のコンテンツを持つXMLファイルを作り、pom.xmlと同じディレクトリに配置します。このファイルは、Oracle Application Container Cloud Serviceへデプロイされるデプロイメント・アセンブリを指定するものです。
<assembly
<id>dist</id>
<formats>
<format>zip</format>
</formats>
<includeBaseDirectory>false</includeBaseDirectory>
<files>
<file>
<source>manifest.json</source>
<outputDirectory></outputDirectory>
</file>
</files>
<fileSets>
<fileSet>
<directory>${project.build.directory}</directory>
<outputDirectory></outputDirectory>
<includes>
<include>${project.artifactId}-${project.version}.jar</include>
</includes>
</fileSet>
</fileSets>
</assembly>
|
To let Application Container cloud know the jar to run once the application is deployed, you need to create a “manifest.json” file specifying the jar name as shown below:
{
"runtime": {
"majorVersion": "8"
},
"command": "java -jar AxonApp-0.0.1-SNAPSHOT.jar",
"release": {},
"notes": "Axon Spring Boot App"
}
|
下図はこのサンプルのプロジェクト構造を示しています。
 |
| Figure 2 Project Structure |
アプリケーションJarファイルは上記のmanifestファイルとともにZipにまとめられ、Application Container Cloud Serviceにデプロイするためにアップロードされます。Spring BootアプリケーションのApplication Container Cloud Serviceへのデプロイについて詳細は以下のURLをご覧ください。
Deploying an Application to Oracle Application Container Cloud Service
http://www.oracle.com/webfolder/technetwork/tutorials/obe/cloud/apaas/deployment/deployment.html
アプリケーションが無事にデプロイ完了したら、以下のURLを使ってCartのサービスを呼び出すことができます。
- http://<host>:<port>/view
- http://<host>:<port>/add/cart/<noOfItems>
- http://<host>:<port>/remove/cart/<noOfItems>
まず RESTエンドポイント "view" を呼び出すと、スタートアップクラスに追加した2つのカートと、追加されたアイテムの数が表示されます。他の2つのRESTサービスを呼び出してCartからアイテムを追加や削除、view RESTサービスを使用して更新された項目数を取得できます。REST呼び出しの結果は、その時点でのCart内のアイテム個数ならびにCartを示すシンプルなJSON構造として取得できます。
Conclusion
このエントリでは、CQRSとイベントソーシングパターンを使用したマイクロサービスアプリケーションの開発について紹介しました。この分野における他の先進的な概念や最新のアップデートについては、以下のリソースを参照してください。
References
