AxonFramework命令分发

使用显式的命令分发机制有许多优点。首先，有一个单独的对象，清楚地描述了客户端的意图。通过记录的命令，你可以存储意图和相关资料备查。命令处理也很容易将你的命令处理组件暴露给远程客户端，例如通过web服务。测试也变得更容易，只需定义起始情境(given)，就可以定义测试脚本，通过列出一些事件和命令，命令去执行(when)并且预期结果(then)。最后一个主要优点是，在同步和异步，以及本地与分布式命令处理之间切换非常容易。

这并不意味着使用显式命令对象进行命令分发是唯一可行的方法。Axon的目标不是规定一种特定的工作方式，而是支持你按照自己的方式去做，同时提供最佳实践作为默认行为。仍有可能使用一个服务层，你可以调用它来执行命令。该方法只需要启动一个工作单元，并在方法完成后执行提交或回滚操作。

下一节提供一个任务的概述，与用Axon Framework设置命令分发基础设施相关。

命令网关

命令网关是有助于命令分发机制的实用接口。虽然使用网关分发命令为非必须，但通常是最简单的选择。

使用命令网关有两种方法。第一个是使用Axon提供的CommandGateway接口和DefaultCommandGateway实现。命令网关提供了一些方法，允许你发送一个命令并同步等待结果，也可以超时或异步等。

另一个选择也许是最灵活的。你使用CommandGatewayFactory可以把几乎任何接口变成一个命令网关。这允许你使用强类型定义你的应用程序的接口，并声明自己的(checked)业务异常。Axon将在运行时自动为该接口生成一个实现。

配置命令网关

你的自定义的和一个由Axon提供的网关，需要至少配置为可以访问命令总线。此外，命令网关可以用一个RetryScheduler、CommandDispatchInterceptor D和CommandCallback S配置。

RetryScheduler能够调度重试当命令执行失败时，IntervalRetryScheduler是一个实现了在指定的时间间隔内，重试给定的命令直到成功，或重试到最大重试次数。当一个命令由于明确非瞬态异常而失败，完全不做重试。注意，重试调度器仅在一个命令由于RuntimeException失败时调用。受检异常被认为“business exception”，不会解发重试。RetryScheduler主要用途是在分布式命令总线上分发命令时候。如果一个节点失败，重试调度器将导致一个命令被分发到能够处理命令的下一个节点。(看Distributing the Command Bus)

CommandDispatchInterceptors允许CommandMessages的修改优先分发它们到命令总线。与配置在CommandBus上的CommandDispatchInterceptors相反，只有在通过网关发送消息时才调用这些拦截器。例如，可以使用拦截器附加元数据到一个命令或做验证。

为每个命令发送调用CommandCallbacks。这就为通过该网关发送的所有命令提供了一些通用行为，而不用管它们的类型。

创建自定义命令网关

Axon允许自定义接口用作命令网关。接口中声明的每个方法的行为是基于参数类型、返回类型和声明的异常。使用此网关不仅方便，而且允许你在需要时模拟你的接口，从而使测试更加容易。

这就是参数如何影响CommandGateway的行为：

• 第一个参数预期是要发送的实际命令对象。
  由@MetaDataValue注解的参数将它们的值分配给元数据字段，其中标识符作为被注解的参数传递。
  元数据类型的参数将与CommandMessage上的元数据合并。如果它们key相等，那么后面的参数定义的元数据将覆盖前面参数的元数据。
• CommandCallback类型的参数在命令被处理后会调用onSuccess或onFailure。你可以传递多个回调，并且可以和返回值相结合。在这种情况下，回调函数的调用将总是与返回值(或exception)匹配。
  最后两个参数可以是long(或int)和TimeUnit类型。在这种情况下，只要这些参数说明，该方法最多会阻塞。方法对超时如何反应取决于异常声明方式(见下文)。注意，如果方法的其他属性的方法防止完全阻塞，超时将永远不会发生。

方法声明的返回值也会影响其行为：

• void返回类型将导致该方法立即返回，除非在想要等待的方法上有其他指示，如超时或已声明的异常。
• 返回类型的未来,CompletionStage CompletableFuture将导致该方法立即返回。您可以使用返回的CompletableFuture来访问命令处理器的结果。异常和方法上的超时声明将被忽略。
• 其他任何返回类型会导致方法阻塞，直到结果可用为止。结果转换为返回类型(如果类型不匹配会导致ClassCastException)。

异常有以下的影响：

• 如果命令处理器（或拦截器）抛出该类型的异常，将抛出任何已声明的受检异常。如果受检异常尚未声明，它被包裹在一个CommandExecutionException中，这是一个RuntimeException。
• 当一个超时发生时，默认行为是从方法返回null。这可以通过声明一个TimeoutException改变。如果声明这个异常，TimeoutException将被抛出。
• 当一个线程在等待结果时被中断，默认行为是返回null。在这种情况下,中断标志被设置在线程上。通过在一个方法上声明InterruptedException，这种行为改为抛出的异常。当异常抛出时中断标志被移除，与java规范一致。
  其他运行时异常可以在方法上声明，但除了对API用户进行说明之外，不会有任何影响。

最后，有可能使用注解：

• 在参数部分指定的，用@MetaDataValue注解的参数的值作为元数据值添加。元数据项的key作为参数提供给注解。
• 用@Timeout注解的方法最多会阻塞到指定的时间。如果方法声明超时参数，这个注解将被忽略。
• 用@Timeout注解的类，将导致这个类中声明的所有方法，最多阻塞到指定的时间。除非他们用自己的@Timeout注解或指定超时参数。

public interface MyGateway {

    // fire and forget
    void sendCommand(MyPayloadType command);

    // method that attaches meta data and will wait for a result for 10 seconds
    @Timeout(value = 10,unit = TimeUnit.SECONDS)
    ReturnValue sendCommandAndWaitForAResult(MyPayloadType command,@MetaDataValue("userId") String userId);

    // alternative that throws exceptions on timeout
    @Timeout(value = 20,unit = TimeUnit.SECONDS)
    ReturnValue sendCommandAndWaitForAResult(MyPayloadType command)
                         throws TimeoutException,InterruptedException;

    // this method will also wait,caller decides how long
    void sendCommandAndWait(MyPayloadType command,long timeout,TimeUnit unit)
                         throws TimeoutException,InterruptedException;
}

// To configure a gateway:
CommandGatewayFactory factory = new CommandGatewayFactory(commandBus);
// note that the commandBus can be obtained from the `Configuration` object returned on `configurer.initialize()`.
MyGateway myGateway = factory.createGateway(MyGateway.class);

作者：勇赴 链接：http://www.jianshu.com/p/55483917b8af 來源：简书 著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!