Skip to main content
Version: 2.4.0

实体聚合根

在使用 Phoenix 框架开发项目时,通常需要借助领域驱动设计(DDD),提取领域实体,其中一个特殊的实体为聚合根(领域外对领域内其他实体的访问都需要通过该聚合根),而Phoenix中的实体聚合根概念就对应DDD中的聚合根。

实体聚合根利用 EventSouring 思想把状态直接存在对象当中,参考Event Souring。在Phoenix当中,单个实体聚合根(同一个聚合根ID)处理请求是单线程的,所以不用担心线程安全问题。当然,如果你希望你的应用是可以横向扩容的,就需要设计好聚合根。

maven依赖

<dependency>
<groupId>com.iquantex</groupId>
<artifactId>phoenix-server-starter</artifactId>
<version>2.4.0</version>
</dependency>

实体聚合根

实体聚合根需要使用@EntityAggregateAnnotation来标记类,服务启动后phoenix会校验定义规范和创建实体聚合根类对象。实体聚合根类需要遵循如下规范:

  1. 聚合根类需要使用 @EntityAggregateAnnotation 注解进行标记。
  2. 聚合根类以及聚合根类中的实体均需实现 Serializable 接口,并定义serialVersionUID。
  3. 聚合根类需要提供无参构造函数。

注意:在聚合根上添加 @EntityAggregateAnnotation 注解时,需要通过 aggregateRootType 指定一个聚合根的类别。用来区分不同的聚合根类,该聚合根类别是全局唯一的。且聚合根ID的长度要小于64个字符。

示例代码

@EntityAggregateAnnotation(aggregateRootType = "BankAccount", idempotentSize = 100, bloomSize = 1000000, snapshotInterval = 100000)
public class BankAccountAggregate implements Serializable {
private static final long serialVersionUID = 6073238164083701075L;
// ... act and on method
}

参数配置

配置项描述类型默认值
aggregateRootType聚合根类型String必填项
surviveTime聚合根生存时间,超过该时间聚合根将被从JVM中淘汰,下一次使用时再重建,减少内存使用longLong.MAX_VALUE
snapshotInterval快照间隔,每隔snapshotInterval条消息打印一次快照,加速聚合根重建,0为关闭long1000
idempotentSize聚合根幂等集合大小,取值应大于零,否则可能会导致启动失败long1000
bloomSize布隆过滤器大小,内存中识别幂等,减少读库判断long100000L
dispatcher选择聚合根的调度者,缓解阻塞问题,支持自定义线程池String"phoenix-dispatcher"

命令处理

实体聚合根中需要提供 act() 方法,用来处理Command消息。一般会产生该领域将会发生的Event事件,通过Event事件修改聚合根状态,也可以直接修改(使用CommandSouring模式)。

对于Command命令和Event事件,Phoenix支持三种协议,详情请参见序列化

  • protobuf
  • protostuff
  • java serializable

实体聚合根中的 act() 方法上需要添加 @CommandHandler 注解进行标识。

示例代码

    @CommandHandler(aggregateRootId = "accountCode", isCommandSourcing = true)
public ActReturn act(AccountCreateCmd createCmd) {
this.account = createCmd.getAccountCode();
this.balanceAmt = createCmd.getBalanceAmt();
String message = String.format("初始化账户代码<%s>, 初始化余额<%s>. ", createCmd.getAccountCode(),createCmd.getBalanceAmt());
return ActReturn.builder().retCode(RetCode.SUCCESS).retMessage(message)
.event(new AccountCreateEvent(createCmd.getAccountCode(), createCmd.getBalanceAmt())).build();
}

参数配置

配置项描述类型默认值
AggregateRootId聚合根id,允许多字段组成并且成员嵌套字段访问aggregateRootId = {event.accountCode, event.name}String[]必填项
enableCreateAggregate是否允许自动创建聚合根booleantrue
idempotentIds幂等id(参考幂等操作段落详解)String[]采用phoenix默认的幂等id
isCommandSourcing是否是command sourcing,默认是event sourcingbooleanfalse

关于支持多聚合根Id并且嵌套的使用方式

class CommandA {
String name;
CommandB commandb;
}

class CommandB {
String age;
}
@CommandHandler(aggregateRootId = {name, commandb.age})
public ActReturn act(A cmd) {
}

ActReturn

act() 方法在处理 Command 命令之后需要返回处理的结果以及一些必要的信息,Phoenix对act方法的返回值做了一层封装,统一放到了ActReturn中。 ActReturn中Event事件将会到达两个地方调用方和on方法处理逻辑当中。

public class ActReturn {
// 处理状态码
private final RetCode retCode;
// 返回消息
@Builder.Default
private final String retMessage = "";
// 返回事件
private final Object event;
}

事件处理

在EventSouring模式下,实体聚合根中需要定义 on() 方法,处理 act() 方法中处理Command命令所产生的Event事件。同时Event事件会进行持久化处理(EventStore),当需要重建聚合根内存状态时,通过EventSourcing进行状态重塑。在重塑时可以通过快照进行加速。快照相关配置请参考:配置详情

on() 方法需要遵循如下规范:

  • on() 方法中不能有IO操作,例如:调用DB操作,调用外部接口
  • on() 方法中不能有随机函数,例如:获取系统当前时间,获取随机数

示例代码

    /**
* 处理账户划拨成功事件
* @param event
*/
public void on(Account.AccountAllocateOkEvent event) {
account = event.getAccountCode();
balanceAmt += event.getAmt();
if (event.getAmt() < 0) {
successTransferOut++;
} else {
successTransferIn++;
}
}

查询操作

Phoenix提供了查询聚合根状态的能力。通过在act()方法上添加 @QueryHandler 注解用来标志该act()方法将会执行一个查询操作。在该act()方法中将需要查询的数据封装进Event事件中,通过ActReturn进行返回。

因查询操作不涉及内存状态的修改,只是对数据进行查询,所以查询操作所产生的的Event事件不会进行持久化操作。

示例代码

    @QueryHandler(aggregateRootId = "accountCode")
public ActReturn act(AccountQueryCmd cmd) {

return ActReturn.builder().retCode(RetCode.SUCCESS).retMessage("查询成功").event(
new AccountQueryEvent(account, balanceAmt, successTransferOut, failTransferOut, successTransferIn))
.build();
}

幂等操作

设计参见Phoenix是如何实现幂等的

Phoenix是消息驱动的框架,消息在传输过程中会存在消息重发的情况,同时上游系统也可能有重试策略(两笔消息,但是业务含义一样)

Phoenix会默认对同一笔消息进行幂等处理,同时也支持用户自定义幂等主键来实现业务幂等,可以利用CommandHandler注解中的idempotentIds声明幂等主键。

幂等ID定义数组的首个元素作为幂等类型固定前缀,从第二个字段开始作为幂等ID

@CommandHandler(aggregateRootId = "accountCode", idempotentIds = {"account", "num"} )
public ActReturn act(AccountCreateCmd createCmd) {}

Phoenix在处理幂等时,通过内存有限的幂等集合和EventStore中事件的唯一索引来保证。通常情况下,有限幂等集合可以保证最新处理消息的唯一性,如果是历史消息的重试会命中EventStore的唯一索引来保证。可以在EntityAggregateAnnotation注解上idempotentSize参数来调整幂等集合的大小。

@EntityAggregateAnnotation(aggregateRootType = "BankAccount", idempotentSize = 1000)
public class BankAccountAggregate implements Serializable {}

幂等集合是有限的,当消息在幂等集合判断重复为false时,Phoenix会在从数据库中查询是否处理过了这个消息。框架为了加速幂等的判断,Phoenix引入了自行淘汰的布隆过滤器,用户可以调整布隆过滤器的大小,来尽量是判重在内存中完成,而不去访问数据库。用户可以通过下面配置来调整布隆过滤器大小。

@EntityAggregateAnnotation(aggregateRootType = "BankAccount", bloomSize = 100000L)
public class BankAccountAggregate implements Serializable {}

注意:布隆过滤器不宜太大,太大会使应用占用内存。每个聚合根对象都有一个布隆过滤器,布隆过滤器数量大小和实际占用内存可以参考下面数据。

布隆过滤器Size占用内存
1000017.982KB
100000179.982KB
10000001797.982KB

布隆过滤器只是为了减少判断幂等时的查库频率,一般推荐设置10000即可,实际测试size=10000即可比较准确的判断最近10W个左右的命令。

聚合根释放

Phoenix针对每个聚合根类型提供了一种淘汰机制,对于长时间不活跃的聚合根(聚合根长时间不处理消息),在指定时间之后进行淘汰。

可以在每个聚合根的 @EntityAggregateAnnotation中增加surviveTime不活跃时间可进行配置,配置如下:

  // surviveTime 聚合根淘汰策略-不活跃时间(毫秒)   默认值:Long.MAX_VALUE
@EntityAggregateAnnotation(aggregateRootType = "BankAccount", surviveTime = 1000 * 60)
public class BankAccountAggregate implements Serializable {
}

聚合根对象被释放不代表该聚合根就不可用,如果有命令触发该聚合根做操作,则会通过EventSouring的机制恢复。释放只是为了释放出JVM内存的占用。

Spring 支持

Phoenix可以和SpringCloud 环境相互打通,比如说聚合根在处理过程中,可以通过Feign调用SpringCloud里的服务。通过将聚合根实例纳入Spring管理,使得可以在聚合根中AutoWired其他Bean,或者在其他Bean中AutoWired聚合根实例。以便实现通过Feign接口调用SpringCloud服务。 使用方式如下:

@Getter
@Setter
@Slf4j
@EntityAggregateAnnotation(aggregateRootType = "BankAccount")
public class BankAccountAggregate implements Serializable {

private static final long serialVersionUID = 6073238164083701075L;

/** 账户代码 */
private String account;

/** 账户余额 */
private double balanceAmt;

/** 成功转出次数 */
private int successTransferOut;

/** 失败转出次数 */
private int failTransferOut;

/** 成功转入次数 */
private int successTransferIn;

/** 第三方服务 */
@Autowired
private transient BankAccountService service;
}

注解配置

从phoenix-server 2.2.4开始,依赖了phoenix-stater的环境下,@EntityAggregateAnnotation支持从启动参数/环境变量/配置文件中读取参数。

配置优先级:启动参数 > 环境变量 > 配置文件 > 代码注解自定义 > 注解默认值

配置项描述类型对应的注解函数
{aggregateRootType}聚合根类型,无法从配置文件中读取StringaggregateRootType
quantex.phoenix.server.entityAggregate.{aggregateRootType}.surviveTimelongsurviveTime
quantex.phoenix.server.entityAggregate.{aggregateRootType}.snapshotIntervallongsnapshotInterval
quantex.phoenix.server.entityAggregate.{aggregateRootType}.idempotentSizelongidempotentSize
quantex.phoenix.server.entityAggregate.{aggregateRootType}.bloomSizelongbloomSize

以yaml配置文件为例:

当使用了@EntityAggregateAnnotation,并且aggregateRootTypeBankAccount。则:

quantex:
phoenix:
server:
entityAggregate:
BankAccount: # {aggregateRootType}
surviveTime: 10
snapshotInterval: 10
idempotentSize: 10
bloomSize: 10

自定义线程池

@EntityAggregateAnnotation注解中,属性dispatcher用来指定线程池。
当聚合根中存在阻塞的线程时,可能会造成其他线程饥饿,用户通过使用其他线程池执行阻塞api来解决这个问题。
Phoenix内提供有默认的"phoenix-dispatcher"线程池和可选择使用的"phoenix-blocking-dispatcher"线程池, 另外用户可以通过自己的需求配置。

phoenix-blocking-dispatcher {
type = Dispatcher
executor = "thread-pool-executor"
thread-pool-executor {
fixed-pool-size = 16
}
throughput = 1
}

自选快照存储数据

Phoenix提供给用户SelectiveSnapshot接口,实现该接口可以自定义快照中存储的数据。

快照中存储的数据应该是聚合根中重要的信息,而不是存储整个聚合根对象,这样做可以避免一些问题(如:聚合根中存在无法序列化的数据或虚拟数据)。

示例代码

使用该功能需要用户自定义一个类,这个类用于存储即将写入到快照中的数据,该类中只需要有属性和序列化接口,且每个属性必须与聚合根中存在的字段对应。

@Data
@Builder
public class SnapshotData implements Serializable {
private static final long serialVersionUID = -5908472083137878870L;
/** 账户代码 */
private String account;
/** 账户余额 */
private double balanceAmt;
/** 成功转出次数 */
private int successTransferOut;
/** 失败转出次数 */
private int failTransferOut;
/** 成功转入次数 */
private int successTransferIn;
}

SelectiveSnapshot接口中必须要实现的两个方法配合使用,能达到自选数据写入快照中或从快照中读出自选数据的效果。

@EntityAggregateAnnotation(
aggregateRootType = "BankAccount",
idempotentSize = 100,
bloomSize = 1000000,
snapshotInterval = 100000)
@Getter
@Setter
@Slf4j
public class BankAccountAggregate implements SelectiveSnapshot<SnapshotData> {

/** 账户代码 */
private String account;
/** 账户余额 */
private double balanceAmt;
/** 成功转出次数 */
private int successTransferOut;
/** 失败转出次数 */
private int failTransferOut;
/** 成功转入次数 */
private int successTransferIn;

@Override
public void loadSnapshot(SnapshotData snapshotData) {
this.setAccount(snapshotData.getAccount());
this.setBalanceAmt(snapshotData.getBalanceAmt());
this.setFailTransferOut(snapshotData.getFailTransferOut());
this.setSuccessTransferIn(snapshotData.getSuccessTransferIn());
this.setSuccessTransferOut(snapshotData.getSuccessTransferOut());
}

@Override
public SnapshotData saveSnapshot() {
return SnapshotData.builder()
.account(getAccount())
.balanceAmt(getBalanceAmt())
.successTransferIn(getSuccessTransferIn())
.failTransferOut(getFailTransferOut())
.successTransferOut(getSuccessTransferOut())
.build();
}
}

序列化

Phoenix传输的Message目前支持Java原生,protobuf,protostuff协议。用户可以根据自己的需要配置使用协议,参考配置。

quantex:
phoenix:
server:
default-serializer: proto_stuff

下面有三种序列化测试结果对比,如果追求性能最高的话,推荐使用protobuf,如果追求开发便利,推荐使用protostuff协议。

简单对象测试:
--------------------------------------------------------------------------------------------
name avgSer minSer maxSer avgDeser minDeser maxDeser size
ProtoStuff-Simple 17046 700 91962900 21797 800 151908700 167
ProtoBuffer-Simple 3275 500 5544700 7674 500 16391100 171
Kryo-Simple 20857 1200 126308700 26969 1500 114516800 181
FastJSON-Simple 16840 2600 82678900 54146 2300 169683700 254
JackJSON-Simple 63640 2000 176485900 28913 2600 107764700 258
Jdk-Simple 32379 3500 93950400 111837 12800 117740300 454
XML-Simple 1491812 939400 136007600 468711 190500 323252400 1104
复杂对象测试:
--------------------------------------------------------------------------------------------
name avgSer minSer maxSer avgDeser minDeser maxDeser size
Kryo-Complex 18133 4500 19233700 31186 4900 90632500 1283
ProtoStuff-Complex 188440 55300 64674300 119845 42800 58609600 58485
ProtoBuffer-Complex 54821 24500 10718900 97838 42600 11946800 58485
Jdk-Complex 60064 23100 56048700 252456 108900 61574600 7146
FastJSON-Complex 439399 235800 245405200 438605 168200 360547200 63531
JackJSON-Complex 411030 161300 266387600 491689 168200 188725200 63531
XML-Complex 8435807 6672800 107200800 2061636 1566100 78898400 76915

异常状态码

在框架启动或运行时,会抛出各种异常,框架收到此类异常,会将异常调用栈、异常信息和异常状态码封装为异常事件。

异常事件的结构如下:


public class ExceptionEvent {
/** 异常调用栈 */
private String errorStack;
/** 异常状态码 */
private int errorCode;
/** 异常内容 */
private String errorDesc;
}

用户可以通过定义特定的on方法,接收异常事件,并对异常事件做处理。这里以计数器聚合根为例做使用说明:

@EntityAggregateAnnotation(
aggregateRootType = "Counter",
idempotentSize = 1000,
snapshotInterval = 10)
public class CounterAggregate implements Serializable {
// 别的on方法

/** 接收异常事件,处理异常事件 */
public void on(Phoenix.ExceptionEvent event) {
log.error("exception event message");
}
}
异常状态码状态码说明异常说明
1000
1001业务代码异常用户代码向框架抛出一个异常,框架会捕获并且封装为此状态码
1002聚合根校验失败框架启动时,会校验聚合根配置的合法性,如果不合法,则会抛出此异常
1003聚合根类型不存在在框架运行时,如果处理事件的聚合根不存在,则会抛出此异常
1004业务代码响应的事件集为空当业务代码处理完后,响应给框架的事件集如果为空,则会抛出此异常
1005业务代码异常用户代码向框架抛出一个异常,框架会捕获并且封装为此状态码
1006获取幂等id失败根据配置的参数,如果框架获取命令的幂等id失败,则会抛出此异常
1007未定义处理该命令的业务代码框架收到一个无法处理的命令,则会抛出此异常
1008创建聚合根失败在运行时,创建聚合根失败,则会抛出此异常
1009未找到聚合根在运行时,处理命令的聚合根不存在,并且配置不可自动创建,则会抛出此异常
1010获取聚合根id失败根据配置的参数,如果框架获取命令的聚合根id失败,则会抛出此异常
1011重复定义了聚合根类型在框架中定义了重复的聚合根类型,则会抛出此异常
1012定义了相同的业务代码在框架中定义了处理相同的命令的业务代码,则会抛出此异常
1013反序列化失败框架在收到MQ的消息后,反序列化失败,则会抛出此异常
1014框架的配置参数有误框架的配置参数校验不通过,则会抛出此异常
1015框架执行数据库操作异常框架执行事件持久化操作失败,则会抛出此异常