<返回更多

DDD与微服务集成的第一战役：客户端重试&服务端幂等

2023-10-30  微信公众号  geekhalo
加入收藏

当一个接口从简单的内部调用升级为远程方法调用（RPC）会面临很多问题，比如：

  1. 本地事务失效。在内部调用时，多个方法通常在同一事务中执行，可以使用本地数据库事务来确保数据的一致性。但是，在远程方法调用中，由于涉及到网络通信，事务的边界会扩展到多个系统之间，因此无法直接使用本地事务。如果远程方法调用出现异常，可能会导致事务提交失败，从而产生数据不一致；
  2. 第三状态影响。网络不确定性可能导致远程调用无法成功获得结果，例如网络连接中断、网络超时等。在这种情况下，客户端无法获得期望的结果，调用会以网络错误或超时的方式结束；
  3. 服务版本兼容性问题。如果服务接口发生变化，客户端和服务端的版本不匹配可能导致调用失败；
  4. 性能问题、可用性问题、安全问题等；

由于涉及的问题比较多，这里重点分析和解决 RPC 调用时的第三状态问题。

1. 什么是第三状态

当一个客户端发起一个RPC请求时，服务端可能会返回不同的状态，包括：

  1. 成功：服务端成功完成了客户端发送的请求，并返回对应的响应结果；
  2. 失败：服务端无法成功处理客户端发送的请求，并返回错误信息或异常；
  3. 超时：调用方无法在规定时间收到服务端的处理结果，所以无法知道请求的最终处理结果；

如下图所示：

DDD与微服务集成的第一战役：客户端重试&服务端幂等图片

一般情况下，调用方在规定时间收到被调用方的返回结果，能够非常明确的知道处理结果是成功还是失败。

当网络或被调用方出问题，就会触发超时，比如下图所示：

DDD与微服务集成的第一战役：客户端重试&服务端幂等图片

如果被调用方异常或者网络发生阻塞，调用方发送的 Request 请求没有被正常处理，那调用方只能在等待若干时间后抛出异常进行流程中断。

又或者如下图所示：

DDD与微服务集成的第一战役：客户端重试&服务端幂等图片

被调用方处理时间过长或者网络发生阻塞，调用方无法在规定时间获得最终结果，也只能触发超时中断。

可见，如果发生 超时 情况，对于处理结果就处于未知状态，这就是所谓的“第三状态”：

  1. 被调用方成功接收到请求并完成了处理；
  2. 被调用方完全没有接收到请求；

在出现第三状态时，在不做任何处理前，根本就无法获取最终的处理结果。在该场景下，最通用的解决方案便是 客户端重试 + 服务端幂等。

  1. 客户端重试。指的是在RPC调用出现超时的情况下，客户端自动重新发送相同的请求。然而，在客户端重试时需要注意避免重复执行有副作用的操作，比如避免重复插入数据;
  2. 服务端幂等。指的是对于相同的输入请求，服务端能够产生相同的结果，而且不会对系统状态造成影响。幂等性是为了应对由于重试等原因导致重复执行的副作用;

通过客户端重试和服务端幂等的方式，可以增加RPC调用的可靠性和数据一致性。客户端重试可以处理网络不稳定、服务端故障等导致的失败情况，而服务端幂等性能保证相同的请求不会重复执行或引起数据不一致的问题。这两个技术结合使用，可以提高分布式系统中RPC调用的健壮性和可靠性。

2. 客户端重试

客户端重试指的是在RPC调用失败的情况下，客户端自动重新发送相同的请求。客户端可以设置重试的次数和时间间隔，直到得到预期的成功响应或达到最大重试次数。客户端重试机制可以弥补网络不稳定性或服务端异常导致的调用失败。然而，在客户端重试时需要注意避免重复执行有副作用的操作，比如避免重复插入数据。此外，还需要合理设置重试策略，避免因频繁的重试导致网络负荷增加或服务端压力过大。

如下图所示：

DDD与微服务集成的第一战役：客户端重试&服务端幂等图片

  1. 第一次获取商品信息时，由于网络异常导致请求超时；
  2. 网络异常触发 retry 机制，重新发起新的请求；
  3. 新请求成功发送至商品服务并获取信息，从而使得流程从异常中恢复，最终正常执行完成；

由此可见，重试的工作原理非常简单。

Spring Retry是Spring Framework提供的一个模块，用于简化和增强应用程序中的重试操作。它提供了一种声明式的方式来处理方法调用的重试，以应对在分布式系统或有限资源环境下可能出现的失败情况。

Spring Retry模块的主要特性包括：

  1. 声明式注解：通过使用注解（例如@Retryable、@Recover等），可以将重试行为与方法关联起来。通过在方法上添加@Retryable注解，可以指定需要进行重试的异常类型，最大重试次数，重试间隔等信息；
  2. 容错策略：Spring Retry提供了多种容错策略，包括简单重试、指数退避重试、固定时间间隔重试等，开发者可以根据具体需求选择合适的策略；
  3. 回退机制：如果重试次数超过限定值仍然失败，Spring Retry可以通过@Recover注解来指定一个回退方法，用于执行备选逻辑或返回默认值；

要使用Spring Retry，需要在项目中引入相应的依赖：

<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
</dependency>

2.1. Retryable

重试行为主要是由 @Retryable 注解完成，通过为目标方法添加注解，将其纳入重试管理，Spring Retry将负责在出现失败情况时自动进行重试。

以下是 @Retryable 注解代码示例：

@Service
public class MyService {

    @Retryable(value = {SomeException.class}, maxAttempts = 3)
    public void doSomething() {
        // 需要进行重试的业务逻辑
    }
}

上述示例中的 doSomething() 方法标记为需要进行重试，在捕获到 SomeException 异常时触发重试，最多重试3次。

下表列出了 @Retryable 注解的所有属性及其说明：

属性名

说明

value

定义重试的异常类型。默认情况下，将捕获所有Throwable类型的异常进行重试。可以通过指定一个或多个异常类来限定重试的异常类型，例如@Retryable(value = {SQLException.class, TimeoutException.class})。

maxAttempts

定义最大重试次数，默认为3次。当达到最大重试次数后，如果仍然失败，则不再进行重试。

backoff

定义重试间隔策略。可以使用@Backoff注解来配置重试间隔的退避策略，包括delay（初始延迟时间，默认为0）和maxDelay（最大延迟时间，默认为无限制）。例如，@Backoff(delay = 1000, maxDelay = 5000)表示初始延迟为1秒，最大延迟为5秒。

delayExpression

定义重试间隔的SpEL表达式。可以使用#lastException表示上一次重试时捕获的异常对象。

multiplier

定义指数退避策略的倍数，默认为1。在使用指数退避策略时，每次重试的间隔时间将是delay * (multiplier ^ (重试次数-1))。

random

定义是否启用随机化延迟。默认为false，即不启用随机化延迟。当设置为true时，重试间隔将随机波动一定的范围。

exceptionExpression

定义一个SpEL表达式，用于决定是否进行重试。该表达式可以使用#result表示目标方法的返回值，#lastException表示上一次重试时捕获的异常对象。如果表达式的结果为true，则进行重试；否则，不进行重试。

include

定义需要包含在重试行为中的异常类型，默认为空数组，表示包括所有异常类型。可以指定一个或多个异常类来明确指定需要处理的异常类型。例如，@Retryable(include = {IOException.class, TimeoutException.class})表示仅包括IOException和TimeoutException异常。

exclude

定义需要排除在重试行为之外的异常类型，默认为空数组，表示排除所有异常类型。可以指定一个或多个异常类来明确排除某些异常类型。例如，@Retryable(exclude = {NullPointerException.class})表示排除NullPointerException异常。

这些属性可以根据具体的需求进行配置，以实现灵活的重试策略。

2.2. Recover

Spring Retry 的 fallback 机制是在重试失败后，执行备选逻辑的一种处理方式。通过使用 @Recover 注解，在重试失败后，可以调用备选逻辑方法来完成错误处理、数据清理等操作。

具体来说，当 @Retryable 注解标记的方法达到最大重试次数或者抛出了无法重试的异常时，Spring Retry将会尝试查找与该方法参数类型相同的方法，并在找到时调用它。

@Retryable(maxAttempts = 3)
public void someMethod(String arg1, int arg2) {
    // 重试业务逻辑
}

@Recover
public void recover(String arg1, int arg2) {
    // 备选逻辑
}

当 someMethod() 方法达到最大重试次数或抛出无法重试的异常时，Spring Retry 将会查找并调用 recover() 方法，并传递相同的方法参数。在 recover() 方法中，应该针对重试失败的情况编写备选逻辑，例如记录日志、发送通知等操作。

需要注意的是，@Recover 注解必须放置在其对应的重试方法所属的类中，并且方法参数类型必须与重试方法一致。如果有多个重试方法，每个重试方法都可以对应一个@Recover方法，以满足不同的备选逻辑需求。

在使用fallback机制时，应该仔细考虑备选逻辑的实现方式，确保其能够正确处理重试失败的情况，并对数据的一致性和完整性产生最小的影响。

2.3. 不同场景下的 Retry 和 Fallback

@Retryable 和 @Recover 都是添加在类方法上的注解，不管是什么场景下的请求只会走固定流程。这样的设计在复杂场景下是否够用？

之前，我们看到通过 Retry 恢复网络抖动的场景；接下来让我们看另一个场景，如下图所示：

DDD与微服务集成的第一战役：客户端重试&服务端幂等图片

image

商品服务流量激增，导致 DB CPU 飙升，出现大量的慢 SQL，这时触发了系统的 Retry 会是怎样的结果？

  1. 在获取商品失败后，系统自动触发 Retry 机制；
  2. 由于是商品服务本身出了问题，第二次请求仍旧失败；
  3. 服务又触发了第三次请求，仍未获取结果；
  4. 达到最大重试次数，仍旧无法获取商品，只能通过异常中断用户请求；

通过 Retry 机制未能将流程从异常中恢复过来，反而给下游的 商品服务 造成了巨大伤害。

  1. 商品服务压力大，响应时间长；
  2. 上游系统由于超时触发自动重试；
  3. 自动重试增大了对商品服务的调用；
  4. 商品服务请求量更大，更难以从故障中恢复；

这就是常说的“读放大”，假设用户验证是否能够购买请求的请求量为 n，那极端情况下 商品服务的请求量为 3n （其中 2n 是由 Retry 机制造成）

此时，最优解不是进行 Retry 而是直接走 Fallback，给下游服务一定的恢复机会。

同样是对商品服务接口（同一个接口）的调用，在不同的场景需要使用不同的策略用以适配不同的业务流程，通常情况下：

  1. Command 场景优先使用 Retry 策略

这种流量即为重要，最好能保障流程的完整性

通常写流量比较小，小范围 Retry 不会对下游系统造成巨大影响

  1. Query 场景优选使用 Fallback 策略

面对不同的业务场景，你会怎么做呢？准备两组不同的方法根据业务场景分别调用？

2.4. SmartFAIlure

SmartFailure 不是为不同的场景使用不同的方法，而是根据请求上下文信息，动态的走 Retry 或 Fallback，从而更好的适应复杂的业务场景。

整体设计如下：

DDD与微服务集成的第一战役：客户端重试&服务端幂等图片

整体流程如下：

  1. 读取配置信息，将请求类型（ActionType）绑定到线程上下文；
  2. 然后执行正常业务逻辑
  3. 当调用 @SmartFault 注解的方法时，会被 SmartFaultMethodInterceptor 拦截器拦截

拦截器通过 ActionTypeProvider 获取当前的 ActionType；

根据 ActionType 对请求进行路由；

如果是 COMMAND 操作，将使用 RetryTemplate 执行请求，在发生异常时，通过重试配置进行请求重发，从而最大限度的获得远程结果；

如果是 QUERY 操作，将使用 FallbackTemplate（重试次数为0的 RetryTemplate）执行请求，当发生异常时，调用 fallback 方法，执行配置的 recover 方法，直接使用返回结果；

  1. 获取远程结果后，执行后续的业务逻辑；
  2. 最后，ActionAspect 将 ActionType 从线程上下文中移除；

使用前需添加 lego 依赖，具体如下：

<dependency>
    <groupId>com.geekhalo.lego</groupId>
    <artifactId>lego-starter</artifactId>
    <version>最新版本</version>
</dependency>
2.4.1. ActionTypeProvider

首先，需要准备一个 ActionTypeProvider 用以提供上下文信息。ActionTypeProvider 接口定义如下：

public interface ActionTypeProvider {
    ActionType get();
}

public enum ActionType {
    COMMAND, QUERY
}

通常情况下，我们使用 ThreadLocal 组件将 ActionType 存储于线程上下文，在使用时从上下中获取相关信息。

public class ActionContext {
    private static final ThreadLocal<ActionType> ACTION_TYPE_THREAD_LOCAL = new ThreadLocal<>();

    public static void set(ActionType actionType){
        ACTION_TYPE_THREAD_LOCAL.set(actionType);
    }

    public static ActionType get(){
        return ACTION_TYPE_THREAD_LOCAL.get();
    }

    public static void clear(){
        ACTION_TYPE_THREAD_LOCAL.remove();
    }
}

有了上下文之后，ActionBasedActionTypeProvider 直接从 Context 中获取 ActionType 具体如下：

@Component
public class ActionBasedActionTypeProvider implements ActionTypeProvider {
    @Override
    public ActionType get() {
        return ActionContext.get();
    }
}

如何对请求进行标记？如何对 ActionType 进行管理（包括信息绑定和信息清理）？最常用的方式便是：

  1. 提供一个注解，在方法上添加注解用于对 ActionType 的配置；
  2. 提供一个拦截器，对方法调用进行拦截：

方法调用前，从注解中获取配置信息并绑定到上下文；

执行具体的业务方法；

方法调用后，主动清理上下文信息；

核心实现为：

@Target({ ElementType.METHOD, ElementType.TYPE })
@Retention(RetentionPolicy.RUNTIME)
public @interface Action {
    ActionType type();
}

@Aspect
@Component
@Order(Integer.MIN_VALUE)
public class ActionAspect {
    @Pointcut("@annotation(com.geekhalo.lego.faultrecovery.smart.Action)")
    public void pointcut() {
    }

    @Around(value = "pointcut()")
    public Object action(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        Action annotation = methodSignature.getMethod().getAnnotation(Action.class);
        ActionContext.set(annotation.type());
        try {
            return joinPoint.proceed();
        }finally {
            ActionContext.clear();
        }
    }
}

在这些组件的帮助下，我们只需在方法上基于 @Action 注解进行标记，便能够将 ActionType 绑定到上下文。

2.4.2. @SmartFault

在完成 ActionType 绑定到上下文之后，接下来要做的便是对 远程接口 进行配置。远程接口的配置工作主要由 @SmartFault 来完成。其核心配置项包括：

配置项

含义

默认配置

recover

fallback 方法名称

 

maxRetry

最大重试次数

3

include

触发重试的异常类型

 

exclude

不需要重新的异常类型

 

测试 Demo 如下：

@Service
@Slf4j
@Getter
public class RetryService3 {
    private int count = 0;

    private int retryCount = 0;
    private int fallbackCount = 0;
    private int recoverCount = 0;

    public void clean(){
        this.retryCount = 0;
        this.fallbackCount = 0;
        this.recoverCount = 0;
    }

    /**
    * Command 请求，启动重试机制
    */
    @Action(type = ActionType.COMMAND)
    @SmartFault(recover = "recover")
    public Long retry(Long input) throws Throwable{
        this.retryCount ++;
        return doSomething(input);
    }

    /**
    * Query 请求，启动Fallback机制
    */
    @Action(type = ActionType.QUERY)
    @SmartFault(recover = "recover")
    public Long fallback(Long input) throws Throwable{
        this.fallbackCount ++;
        return doSomething(input);
    }

    @Recover
    public Long recover(Throwable e, Long input){
        this.recoverCount ++;
        log.info("recover-{}", input);
        return input;
    }

    private Long doSomething(Long input) {
        // 偶数抛出异常
        if (count ++ % 2 == 0){
            log.info("Error-{}", input);
            throw new RuntimeException();
        }
        log.info("Success-{}", input);
        return input;
    }
}
2.4.3. 测试

最后，对代码进行简单测试：

@SpringBootTest(classes = DemoApplication.class)
public class RetryService3Test {
    @Autowired
    private RetryService3 retryService;

    @BeforeEach
    public void setup(){
        retryService.clean();
    }

    @Test
    public void retry() throws Throwable{
        for (int i = 0; i < 100; i++){
            retryService.retry(i + 0L);
        }

        Assertions.assertTrue(retryService.getRetryCount() > 0);
        Assertions.assertTrue(retryService.getRecoverCount() == 0);
        Assertions.assertTrue(retryService.getFallbackCount() == 0);
    }

    @Test
    public void fallback() throws Throwable{
        for (int i = 0; i < 100; i++){
            retryService.fallback(i + 0L);
        }

        Assertions.assertTrue(retryService.getRetryCount() == 0);
        Assertions.assertTrue(retryService.getRecoverCount() > 0);
        Assertions.assertTrue(retryService.getFallbackCount() > 0);
    }
}

运行 retry 测试，日志如下：

[main] c.g.l.c.f.smart.SmartFaultExecutor       : action type is COMMAND
[main] c.g.l.faultrecovery.smart.RetryService3  : Error-0
[main] c.g.l.c.f.smart.SmartFaultExecutor       : Retry method public JAVA.lang.Long com.geekhalo.lego.faultrecovery.smart.RetryService3.retry(java.lang.Long) throws java.lang.Throwable use [0]
[main] c.g.l.faultrecovery.smart.RetryService3  : Success-0

可见，当 action type 为 COMMAND 时：

方法主动进行重试，流程从异常中恢复，处理过程和效果符合预期。

运行 fallback 测试，日志如下：

[main] c.g.l.c.f.smart.SmartFaultExecutor       : action type is QUERY
[main] c.g.l.faultrecovery.smart.RetryService3  : Error-0
[main] c.g.l.c.f.smart.SmartFaultExecutor       : recover From ERROR for method ReflectiveMethodInvocation: public java.lang.Long com.geekhalo.lego.faultrecovery.smart.RetryService3.fallback(java.lang.Long) throws java.lang.Throwable; target is of class [com.geekhalo.lego.faultrecovery.smart.RetryService3]
[main] c.g.l.faultrecovery.smart.RetryService3  : recover-0

可见，当 action type 为 QUERY 时：

异常后自动执行 fallback，将流程从异常中恢复过来，处理过程和效果符合预期。

3. 服务端幂等

服务端幂等指的是对于相同的输入请求，服务端能够产生相同的结果，而且不会对系统状态造成影响。幂等性是为了防止由于重试等原因导致的重复执行带来的副作用。在RPC中，服务端需要设计和实现幂等性的方法，确保多次接收到相同的请求时能正确处理，而不会重复执行对系统状态有改变的操作。

3.1. 幂等与幂等接口

幂等是指对同一个操作的多次执行所产生的影响与一次执行的影响相同，即不会因为多次执行而产生额外的副作用。在分布式系统中，由于网络等各种因素的存在，可能会导致对同一个操作进行多次执行，此时如果该操作是幂等的，那么就可以避免数据冲突和重复处理问题。

幂等接口是指对外提供的接口，它们所提供的业务操作具有幂等性。也就是说，在客户端多次调用该接口时，每次调用都会产生相同的结果，并且不会产生额外的副作用。

例如，银行转账操作就应该是一个幂等接口。假设客户端已经成功地向银行发起了一次转账请求，但由于网络不稳定等原因，导致该请求的响应没有及时返回给客户端。此时客户端可能会误以为转账请求失败，进而再次发送同样的转账请求。如果银行的转账接口是幂等的，那么无论客户端发送多少次转账请求，最终的结果都应该是相同的——只有一次转账操作会被执行，其他的请求都会被忽略。

3.2. 天然幂接口

有些接口天然就具备幂等性。这些接口通常是对资源的查询操作，不会对资源进行修改。以下是一些天然具备幂等性的场景：

这些接口在设计上更容易满足幂等性的要求，因为它们的操作本身并没有产生副作用，不会对数据进行修改。在分布式系统中，可以安全地多次调用这些接口，而不会引发数据冲突。

3.3. 非幂等接口

非幂等接口是指对资源进行修改、状态进行变更而产生副作用的接口，多次调用可能会导致不同的结果。以下是一些常见的非幂等接口：

这些非幂等接口在设计上需要特别注意，并采取合适的措施来确保数据的一致性和操作的正确性。

3.4. 重复请求的处理方式

当系统检测出当前请求是重复请求时，通常会有两种处理策略：

这两种方案在实现上差异不大，但在客户端使用中差异巨大。

所以，在幂等设计中，优先使用 “直接返回上次请求结果” 方案。

综上分析，幂等可以做为一种通用能力与业务处理逻辑进行充分解构，这就是组件封装的前提。

3.5. 幂等组件

我们可以将幂等封装成一种能力，然后在需要幂等保护的业务方法上进行配置，从而实现幂等能力与业务方法的彻底解耦。

整体设计如下图所示：

DDD与微服务集成的第一战役：客户端重试&服务端幂等图片

整体设计比较简单，运行流程如下：

从设计上看，系统中可以同时配置多个 IdempotentExecutorFactory，然后根据不同的业务场景设置不同的 executorFactory。

幂等处理的核心流程如下：

DDD与微服务集成的第一战役：客户端重试&服务端幂等图片

IdempotentExecutor 处理核心流程如下：

使用前需要引入 lego starter，在 maven pom 中添加如下信息：

<groupId>com.geekhalo.lego</groupId>
<artifactId>lego-starter</artifactId>
<version>最新版本</version>
3.5.1. 配置 dbExecutorFactory

以 JpaRepository 为例实现对 IdempotentExecutorFactory 的配置，具体如下：

@Configuration
public class IdempotentConfiguration extends IdempotentConfigurationSupport {

    @Bean("dbExecutorFactory")
    public IdempotentExecutorFactory dbExecutorFactory(JpaBasedExecutionRecordRepository recordRepository){
        return createExecutorFactory(recordRepository);
    }
}

其中，IdempotentConfigurationSupport 已经提供 idempotent 所需的很多 Bean，同时提供 createExecutorFactory(repository) 方法，用以完成 IdempotentExecutorFactory 的创建。

使用 Jpa 需要调整 EnableJpaRepositories 相关配置，具体如下：

@Configuration
@EnableJpaRepositories(basePackages = {
        "com.geekhalo.lego.core.idempotent.support.repository"
}, repositoryFactoryBeanClass = JpaBasedQueryObjectRepositoryFactoryBean.class)
public class SpringDataJpaConfiguration {
}

其中，com.geekhalo.lego.core.idempotent.support.repository 为固定包名，指向 Jpa 默认实现 JpaBasedExecutionRecordRepository，Spring Data Jpa 会自动生成实现的代理对象。

最后，在数据库中增加 幂等所需表，sql 如下：

CREATE TABLE `idempotent_execution_record` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
   `type` int(11) NOT NULL,
   `unique_key` varchar(64) NOT NULL,
   `status` int(11) NOT NULL,
   `result` varchar(1024) DEFAULT NULL,
   `create_date` datetime DEFAULT NULL,
   `update_date` datetime DEFAULT NULL,
   PRIMARY KEY (`id`),
   UNIQUE KEY `unq_type_key` (`type`,`unique_key`)
) ENGINE=InnoDB;

至此，便完成了基本配置。

【注】关于 Spring data jpa 配置，可以自行到网上进行检索。

3.5.2. 幂等保护示例

在方法上增加 @Idempotent 注解便可以使其具备幂等保护，示例如下：

@Idempotent(executorFactory = "dbExecutorFactory", group = 1, keyEl = "#key",
        handleType = IdempotentHandleType.RESULT)
@Transactional
public Long putForResult(String key, Long data){
    return put(key, data);
}

其中 @Idempotent 为核心配置，详细信息如下：

RESULT，直接返回上次的执行结果

ERROR，直接抛出 RepeatedSubmitException 异常

编写简单的测试用例如下：

@Test
void putForResult() {
    BaseIdempotentService idempotentService = getIdempotentService();
    String key = String.valueOf(RandomUtils.nextLong());
    Long value = RandomUtils.nextLong();

    {   // 第一次操作，返回值和最终结果符合预期
        Long result = idempotentService.putForResult(key, value);
        Assertions.assertEquals(value, result);
        Assertions.assertEquals(value, idempotentService.getValue(key));
    }

    {   // 第二次操作，返回值和最终结果 与第一次一致（直接获取返回值，没有执行业务逻辑）
        Long valueNew = RandomUtils.nextLong();
        Long result = idempotentService.putForResult(key, valueNew);

        Assertions.assertEquals(value, result);
        Assertions.assertEquals(value, idempotentService.getValue(key));
    }
}

运行测试用例，测试通过，可得出如下结论：

这是最常见的一种工作模式，除直接返回上次执行结果外，当发生重复提交时也可以抛出异常中断流程，只需将 handleType 设置为 ERROR 即可，具体如下：

@Idempotent(executorFactory = "dbExecutorFactory", group = 1, keyEl = "#key",
    handleType = IdempotentHandleType.ERROR)
@Transactional
public Long putForError(String key, Long data){
    return put(key, data);
}

编写测试用例，具体如下：

@Test
void putForError() {
    BaseIdempotentService idempotentService = getIdempotentService();
    String key = String.valueOf(RandomUtils.nextLong());
    Long value = RandomUtils.nextLong();

    { // 第一次操作，返回值和最终结果符合预期
        Long result = idempotentService.putForError(key, value);
        Assertions.assertEquals(value, result);
        Assertions.assertEquals(value, idempotentService.getValue(key));
    }

    { // 第二次操作，直接抛出异常，结果与第一次一致
        Assertions.assertThrows(RepeatedSubmitException.class, () ->{
            Long valueNew = RandomUtils.nextLong();
            idempotentService.putForError(key, valueNew);
        });
        Assertions.assertEquals(value, idempotentService.getValue(key));
    }
}

运行测试用例，测试通过，可以得出：

3.5.3. 幂等与异常

异常是一种特殊的返回值！！！

如果将异常看做是一种特殊的返回值，那幂等接口在第二次请求时同样需要抛出异常，示例代码如下：

@Idempotent(executorFactory = "dbExecutorFactory", group = 1, keyEl = "#key",
        handleType = IdempotentHandleType.RESULT)
@Transactional
public Long putExceptionForResult(String key, Long data) {
    return putException(key, data);
}
protected Long putException(String key, Long data){
    this.data.put(key, data);
    throw new IdempotentTestException();
}

@Idempotent 注解没有变化，只是在 putException 方法执行后抛出 IdempotentTestException 异常。

编写简单测试用例如下：

@Test
void putExceptionForResult(){
    BaseIdempotentService idempotentService = getIdempotentService();
    String key = String.valueOf(RandomUtils.nextLong());
    Long value = RandomUtils.nextLong();

    {   // 第一次操作，抛出异常
        Assertions.assertThrows(IdempotentTestException.class,
                ()->idempotentService.putExceptionForResult(key, value));
        Assertions.assertEquals(value, idempotentService.getValue(key));
    }

    {   // 第二次操作，返回值和最终结果 与第一一致（直接获取返回值，没有执行业务逻辑）
        Long valueNew = RandomUtils.nextLong();
        Assertions.assertThrows(IdempotentTestException.class,
                ()->idempotentService.putExceptionForResult(key, valueNew));
        Assertions.assertEquals(value, idempotentService.getValue(key));
    }
}

运行测试用例，用例通过，可知：

3.5.4. 并发保护

如果上一个请求执行尚未结束，新的请求已经开启，那会如何？

这就是最常见的并发场景，idempotent 对其也进行了支持，当出现并发请求时会直接抛出 ConcurrentRequestException，用于中断处理。

首先，使用 sleep 模拟一个耗时的方法，具体如下：

@Idempotent(executorFactory = "dbExecutorFactory", group = 1, keyEl = "#key",
            handleType = IdempotentHandleType.RESULT)
@Transactional
public Long putWaitForResult(String key, Long data) {
    return putForWait(key, data);
}
protected Long putForWait(String key, Long data){
    try {
        TimeUnit.SECONDS.sleep(3);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return put(key, data);
}

putWaitForResult 方法调用时会主动 sleep 3 秒，然后才执行真正的逻辑。

编写测试代码如下：

@Test
void putWaitForResult(){
    String key = String.valueOf(RandomUtils.nextLong());
    Long value = RandomUtils.nextLong();

    // 主线程抛出 ConcurrentRequestException 
    Assertions.assertThrows(ConcurrentRequestException.class, () ->
        testForConcurrent(baseIdempotentService ->
            baseIdempotentService.putWaitForResult(key, value))
    );
}
private void testForConcurrent(Consumer<BaseIdempotentService> consumer) throws InterruptedException {
    // 启动一个线程执行任务，模拟并发场景
    Thread thread = new Thread(() -> consumer.accept(getIdempotentService()));
    thread.start();

    // 主线程 sleep 1 秒，与异步线程并行执行任务
    TimeUnit.SECONDS.sleep(1);
    consumer.accept(getIdempotentService());
}

运行单元测试，测试通过，核心测试逻辑如下：

3.5.5. redis 支持

DB 具有非常好的一致性，但性能存在一定的问题。在一致性要求不高，性能要求高的场景，可以使用 Redis 作为 ExecutionRecord 的存储引擎。

引入 redis 非常简单，大致分两步：

添加 redisExecutorFactory Bean，具体如下：

@Configuration
public class IdempotentConfiguration extends IdempotentConfigurationSupport {

    @Bean("redisExecutorFactory")
    public IdempotentExecutorFactory redisExecutorFactory(ExecutionRecordRepository executionRecordRepository){
        return createExecutorFactory(executionRecordRepository);
    }

    @Bean
    public ExecutionRecordRepository executionRecordRepository(RedisTemplate<String, ExecutionRecord> recordRedisTemplate){
        return new RedisBasedExecutionRecordRepository("ide-%s-%s", Duration.ofDays(7), recordRedisTemplate);
    }

    @Bean
    public RedisTemplate<String, ExecutionRecord> recordRedisTemplate(RedisConnectionFactory redisConnectionFactory){
        RedisTemplate<String, ExecutionRecord> redisTemplate = new RedisTemplate();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.configure(FAIL_ON_UNKNOWN_PROPERTIES, false);
        Jackson2JsonRedisSerializer<ExecutionRecord> executionRecordJackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(ExecutionRecord.class);
        executionRecordJackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        redisTemplate.setValueSerializer(executionRecordJackson2JsonRedisSerializer);
        return redisTemplate;
    }
}

@Idempotent 注解调整如下：

@Idempotent(executorFactory = "redisExecutorFactory", group = 1, keyEl = "#key",
        handleType = IdempotentHandleType.RESULT)
@Override
public Long putForResult(String key, Long data){
    return put(key, data);
}

这样，所有的幂等信息都会存储在 redis 中。

【注】一般 redis 不会对数据进行持久存储，只能保障在一段时间内的幂等性，超出时间后，由于 key 被自动清理，幂等将不再生效。对于业务场景不太严格但性能要求较高的场景才可使用，比如为过滤系统中由于 retry 机制造成的重复请求。

4. 小结

当系统开启微服务化后，服务调用的第三状态就成为了不可回避的话题。通常情况下，会综合使用 客户端重试 和 服务端幂等 两个方案来解决：

写请求，建设使用 Retry 机制，保障最重要的流量不被浪费

读请求，建议使用 Fallback 机制，避免重试流量对下游服务造成巨大压力

以上两种场景，lego 对其都进行了封装，可以方便的应用于业务系统，从而降低微服务的伤害。

关键词：DDD      点击(6)
声明：本站部分内容来自互联网,如有版权侵犯或其他问题请与我们联系，我们将立即删除或处理。
▍相关推荐
更多DDD相关>>>
Copyright @ 简易百科 V 2.0 京ICP备14042104号