1. public class ToDoItemRunner {
3. private CommandGateway commandGateway;
5. public ToDoItemRunner(CommandGateway commandGateway) {
6. this.commandGateway = commandGateway;
7. }
9. public static void main(String[] args) {
10. // 初始化和注入spring bean
11. ApplicationContext applicationContext = new ClassPathXmlApplicationContext("sampleContext.xml");
12. ToDoItemRunner runner = new ToDoItemRunner(applicationContext.getBean(CommandGateway.class));
13. // 开跑
14. runner.run();
15. }
17. private void run() {
18. // 创建待办事项
19. final String itemId = UUID.randomUUID().toString();
20. commandGateway.send(new CreateToDoItemCommand(itemId, "8点要上班"));
21. // 到公司以后
22. commandGateway.send(new MarkCompletedCommand(itemId));
23. }
24. }

1. 客户端创建了一个命令
2. 通过命令网关，向命令总线发送了命令
3. 命令处理器执行了命令，操作了业务实体，持久化
4. 命令处理器根据操作结果产生了事件
5. 事件被发送到事件总线，并被保存起来
6. 若干个订阅过该事件的监听器收到了事件，交给事件处理器实现了事件会造成的后续影响

1. public class ToDoItem {
3. private String id;
4. private String userId;
5. private String commander;
6. private String remark;
7. private Date expectCompleteDate;
8. private int status;
9. ...
10. }
12. public class ToDoItemServiceImpl implements ToDoItemService {
13. private TodoItemDao toDoItemDao;
14. private EmailService emailService;
15. private PointService pointService;
17. public void createToDoItem(String id, String remark, String commander) {
18. ToDoItem toDoItem = new ToDoItem();
19. toDoItem.setId(id);
20. toDoItem.setRemark(remark);
21. toDoItem.setUserId(userId);
22. toDoItem.setCommander(commander);
23. // 一系列时间计算
24. ...
25. toDoItem.setExpectCompleteDate(expectCompleteDate);
26. toDoItemDao.save(toDoItem);
27. }
29. public void changeToDoItemStatus(String id, int status) {
30. ToDoItem toDoItem = toDoItemDao.get(id);
31. toDoItemDao.setStatus(status);
32. toDoItemDao.update(toDoItem);
34. if(status == ToDoItem.STATUS_COMPLETE) {
35. emailService.send("这是邮件内容");
37. if(toDoItem.getExpectCompleteDate.before(new Date())) {
38. pointService.subtract(toDoItem.getUserId(), 1);
39. }
40. }
41. }
42. }

1. public class ToDoItemRunner {
3. private CommandGateway commandGateway;
5. public ToDoItemRunner(CommandGateway commandGateway) {
6. this.commandGateway = commandGateway;
7. }
9. public static void main(String[] args) {
10. ApplicationContext applicationContext = new ClassPathXmlApplicationContext("sampleContext.xml");
11. ToDoItemRunner runner = new ToDoItemRunner(applicationContext.getBean(CommandGateway.class));
12. runner.run();
13. }
15. private void run() {
16. // 创建待办事项
17. final String itemId = UUID.randomUUID().toString();
18. commandGateway.send(new CreateToDoItemCommand(itemId, "8点要上班"));
19. // 到公司以后
20. commandGateway.send(new MarkCompletedCommand(itemId));
21. }
22. }

1. public class ToDoItem extends AbstractAnnotatedAggregateRoot {
3. private String id;
4. private String userId;
5. private String commander;
6. private String remark;
7. private Date expectCompleteDate;
8. private int status;
10. public final static Integer STATUS_UN_COMPLETE = 1;
11. public final static Integer STATUS_COMPLETE = 2;
13. @CommandHandler
14. public ToDoItem(CreateToDoItemCommand command) {
16. // 根据命令创建实体
17. setId(command.getId());
18. setRemark(command.getRemark());
19. setStatus(ToDoItem.STATUS_UN_COMPLETE);
21. // 事件产生
22. apply(new ToDoItemCreatedEvent(this);
23. }
25. @CommandHandler
26. public void markCompleted(MarkCompletedCommand command) {
27. // 超没超时，是两个不同的结果事件
28. if(toDoItem.getExpectCompleteDate.before(new Date())) {
29. apply(new ToDoItemTimeOutCompletedEvent(id));
30. } else {
31. apply(new ToDoItemCompletedEvent(id));
32. }
33. }
35. @EventHandler
36. public void onCreate(ToDoItemCreatedEvent event) {
37. System.out.println("待办事项" + event.getToDoItem().getRemark() + "创建成功");
38. }
40. @EventHandler
41. public void onMarkComplete(ToDoItemCompletedEvent event) {
42. System.out.println("待办事项" + event.getId() + "已完成");
43. }
45. /**
46. / 一个事件处理器负责扣分
47. */
48. @EventHandler
49. public void onTimeOutMarkCompleteForPoint(ToDoTimeOutItemCompletedEvent event) {
50. commandGateway.send(new SubtractUserPointCommand(itemId, 1));
51. System.out.println("待办事项" + event.getId() + "超时完成");
52. }
54. /**
55. / 另一个事件处理器负责发邮件
56. */
57. @EventHandler
58. public void onTimeOutMarkCompleteForEmail(ToDoTimeOutItemCompletedEvent event) {
59. commandGateway.send(new SendEmailCommand(itemId, "这是邮件内容"));
60. System.out.println("待办事项" + event.getId() + "超时完成");
61. }
62. }

如上，Axon是为了业务需求复杂、高性能、高可扩展的系统而生的。对那些几张小表CRUD以后也不会再大动的系统，就不适用了。后面的章节里，我们再慢慢看Axon的各个组件都提供了些什么样的特性，来完成业务的处理。

• 由于对象不实时保存，所以要有不依赖于数据库ACID的新的灾备机制。从事件仓储中取出这个对象的操作事件流，一个一个进行重放。new一个新的对象，通过重放事件的方法，把所有的历史事件再解析一遍，最终对象又回到了宕机前的状态。
  
• 内存是有限的，对象过多时终究要借助硬盘，所以应该借助一种有冷热数据自动清理机制的缓存来保存内存中对象，内存满时，清理掉使用率低的对象，当对象不存在时，可以从数据源重新读取。
  
• 内存中的对象是无法进行复杂条件查询的，因此还需要有一个区别于实体持久数据源的查询数据库，专门支持查询操作。由于在高并发修改的情况下，对象内容变得实在太频繁了，所以没必要每个修改都去update查询数据库，如果一秒钟会发生几次几十次变化，实时查询还有实际意义吗？一次网络点击期间都不知发生多少变化了。更实际的做法是监听一下对象操作所产生的事件，在一定的时机去更新一次查询数据库就可以了。
  
• 查询数据库可以不遵循实体结构，怎么样便于查询优化就怎么设计
  
• 仅建议对高频率修改的实体进行Event Sourcing维护，如秒杀商品库存，金融交易频繁的实时帐户余额变动等。一般情况还是会采用传统方式，后面的框架介绍例子也会结合传统方式去说明。实时
  

2. // 创建一个实例：
3. GatewayProxyFactory factory = new GatewayProxyFactory(commandBus);
4. MyGateway myGateway = factory.createInstance(MyGateway.class);
6. public interface MyGateway {
8. // 调了就不管了，非堵塞
9. void sendCommand(MyPayloadType command);
11. // 方法附加了一个元数据userId到命令上，并且会堵塞等待方法执行，10秒钟不返回就超时，并返回Null
12. @Timeout(value = https://www.it610.com/article/10, unit = TimeUnit.SECONDS)
13. ReturnValue sendCommandAndWaitForAResult(MyPayloadType command,
14. @MetaData("userId") String userId);
16. // 如果超时或者线程中断，会抛出相应异常
17. @Timeout(value = https://www.it610.com/article/20, unit = TimeUnit.SECONDS)
18. ReturnValue sendCommandAndWaitForAResult(MyPayloadType command)
19. throws TimeoutException, InterruptedException;
21. // 这个方法也会堵塞等待结果，等多久由调用方决定
22. void sendCommandAndWait(MyPayloadType command, long timeout, TimeUnit unit)
23. throws TimeoutException, InterruptedException;
24. }

• 第一个参数必须是你期望发送的实际命令对象。
  
• 用@MetaData注解的参数对象值，会被加到命令的元数据字段上，比如来源客户端时间戳之类的。
  
• 参数上的元数据会和CommandMessage里的元数据合并，对于key相同的元数据值，后定义的会覆盖掉先定义的。就是说，如果你的命令对象里原来已经有一个userId字段，这里又定义了一个，那这个就会覆盖掉原来的。
  
• CommandCallback类型的参数，会在命令处理完以后，被调用它的onSuccess或onFailure方法。你还可以通过回调类传递更多处理结果的信息，比如你自己的类和异常。在想要并行处理多个命令时，你可以通过这个类非堵塞拿到运行结果。
  
• 最后两个参数可能是个long或int值，以及TimeUnit时间单位枚举。如果加了这两个，方法就会堵塞到参数设定的超时时间结束为止。至于超时以后会发生什么，取决于你定义在方法上的异常，下面再解释。注意如果其它属性完全阻止了堵塞等待，这两参数就无效了。
  

• 返回类型为void的方法会立即返回，不管是不是还有其他看起来会堵塞的参数比如抛出异常、超时时间等。
  
• Future返回类型的方法也会立即返回，可以从Future中拿到执行结果，抛出的异常和超时时间等仍然会被忽略。
  
• 任何其他类型的返回值都会造成方法调用堵塞直到返回结果，Axon会帮你把返回值转成你定义的类，不匹配的话可能会抛出ClassCastException。
  

• 在方法上定义的受检查的异常只要命令处理器或者拦截器一抛出，Axon就会直接在方法上抛出给你自己处理。如果被抛出的异常你没有定义在接口方法上，Axon会把它包装成一个CommandExecutionException，这是一个RuntimeException
  
• 在超时的时候，默认情况方法会返回Null，但是如果你在方法上定义了一个TimeoutException，那超时时就会给你抛出TimeoutException。
  
• 如果在等待结果的时候线程中断了，默认也会返回Null，中断标识会设置在线程中。在方法上定义一个InterruptedException可以改变方法行为。
  
• 其他的运行时异常也可以定义在方法上以便捕获，但是不会对行为有任何影响的。
  最后是注解的影响：
  
• 加在参数上的@MetaData注解会把这个参数做为元数据键值对的value，而key可以写成注解的参数，如@MetaData("userId") String userId。
  
• 如果在方法上加了@Timeout会导致方法堵塞直到超时，但如果参数里有一个timeout参数，这个注解会被忽略。
  
• 如果给整个接口类加上@Timeout会导致所有方法都堵塞，除非方法上有@Timeout或者参数里有超时参数来覆盖掉它的影响。

1. public interface CommandGateway {
3. void send(Object command, CommandCallback callback);
5. R sendAndWait(Object command);
7. R sendAndWait(Object command, long timeout, TimeUnit unit);
9. void send(Object command);
10. }
12. ...
14. // 同步调用
15. ToDoItem toDoItem = commandGateway.sendAndWait(new MarkCompletedCommand(itemId1), 15, TimeUnit.SECONDS);
17. // 并发调用
18. FutureCallback future1 = new FutureCallback();
19. FutureCallback future2 = new FutureCallback();
20. commandGateway.send(new MarkCompletedCommand(itemId1), future1);
21. commandGateway.send(new MarkCompletedCommand(itemId1), future2);
23. ToDoItem toDoItem1 = future1.get();
24. ToDoItem toDoItem2 = future2.get();

前面的命令网关，就是在这个的基础上进行了包装，增加了更灵活多样化的调用选项。

关于高大上的Disruptor技术，你可以看这里的中文介绍。
简单地描述，Disruptor利用数组提供了一个环状的集合，而不是队列，数据被不停地加入这个环里，一个指针不断地从头到尾去遍历获取，处理好了再塞回到环里等待下一次处理，无穷无尽，还充分利用了CPU的设计机制，性能非常好，至于为什么好，看上面的介绍吧。

• DisruptorCommandBus只支持以事件驱动方式维护的聚合实体，不适合用传统的“取-改-存”方式维护的聚合实体。命令总线也只是为这些由Disruptor处理的聚合实体提供了一个仓储(Repository)的作用。想要拿到一个仓储的引用，可以用createRepository(AggregateFactory)方法。
  
• 一个命令只能导致一个聚合实体发生状态变化，即不能用一个命令改变两个无关实体的状态值。
  
• 如果启用了缓存的话，你不能让两个不同类型的聚合实体拥有相同的id标识，比如id为1的User和id为1的Order不能同时存在。
  
• 用它来处理的命令通常不应该在工作单元里产生一个会回滚的错误，一旦发生了回滚，DisruptorCommandBus就没法保证你的命令会按分发的顺序被处理了。另外回滚后的重试也会消耗它的计算性能。发生了错误以后应该通过后续的处理措施来弥补而不是回滚。
  
• 如果你刚创建了一个新的聚合实体的实例，那些已经发出去，排着队要修改它的命令可能没办法精确地按它们分发的顺序去修改你的新实例。如果这些修改命令要修改的是一个已经存在的实体，顺序就不会有问题。为了保证顺序，在创建新实体的那个命令上用一个回调，等待得到成功结果以后，再发送那些修改命令，也就等几毫秒的事。不要把对同一个新实体的创建和更新命令一次全给发出去。
  配置DisruptorCommandBus
  要构建一个DisruptorCommandBus实例，你需要一个AggregateFactory聚合工厂，一个事件总线EventBus还有一个EventStore事件仓储。这些组件的详细介绍看第5章和第6.1章。最后你还需要一个CommandTargetResolver接口，按名字理解，命令目标解析器，这个东西是为了告诉disruptor你的命令想要操作的是哪一个聚合实体。
  Axon提供了两个实现了这个接口的类，一个是AnnotationCommandTargetResolver是通过分析注解来得知命令的目标实体的，还有一个是MetaDataCommandTargetResolver，是通过分析命令的元数据来得知目标实体的。
  可选的，你可以提供一个自己的DisruptorConfiguration实例来改变默认配置，针对你的应用环境进行进一步的性能调优。在Spring里配是最简单的配置DisruptorCommandBus的方式。
  

2. xmlns:axon="http://www.axonframework.org/schema/core"
3. xmlns="http://www.springframework.org/schema/beans"
4. xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.axonframework.org/schema/core http://www.axonframework.org/schema/axon-core.xsd">
9. aggregate-type="org.axonframework.quickstart.annotated.ToDoItem"
10. command-bus="commandBus"/>
21. repository="toDoRepository"
22. command-bus="commandBus"/>