您当前的位置:首页 > 电脑百科 > 数据库 > Redis

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

时间:2023-10-30 14:12:38  来源:微信公众号  作者: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})表示仅包括IOExceptionTimeoutException异常。

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 策略
  • 大多数展示场景,哪怕部分信息没有获取到对整体的影响也比较小

  • 通常读场景流量较高,Retry 对下游系统的伤害不容忽视

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

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 时:

  • 第一次调用时,触发异常,打印:Error-0
  • 此时 SmartFaultExecutor 主动进行重试,打印:Retry method xxxx
  • 方法重试成功,RetryService3 打印:Success-0

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

运行 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 时:

  • 第一次调用时,触发异常,打印:Error-0
  • SmartFaultExecutor 执行 Fallback 策略,打印:recover From ERROR for method xxxx
  • 调用RetryService3的 recover 方法,获取最终返回值。RetryService3 打印:recover-0

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

3. 服务端幂等

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

3.1. 幂等与幂等接口

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

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

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

3.2. 天然幂接口

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

  • 查询接口:对于只用于查询数据的接口,例如获取用户信息、获取订单详情等,多次调用不会对数据产生影响;
  • 获取接口:对于获取资源的接口,例如获取图片、获取文件等,无论调用多少次,得到的都是相同的资源副本;
  • 计算接口:对于纯计算的接口,例如加法、乘法等数学运算接口,多次调用得到的结果是相同的;
  • 验证接口:对于验证某个条件的接口,例如检查用户名是否已存在、验证手机号是否有效等,多次调用得到的验证结果都是一致的;

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

3.3. 非幂等接口

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

  • 创建资源接口:例如创建用户、创建订单等操作,多次调用会生成多个相同的资源实例,每次调用都会产生不同的结果;
  • 修改资源接口:例如更新用户信息、修改文章内容等操作,多次调用会对同一个资源进行多次修改,每次修改都会产生不同的结果;
  • 删除资源接口:例如删除文件、删除订单等操作,多次调用会多次删除同一个资源,每次调用都会产生不同的结果;

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

3.4. 重复请求的处理方式

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

  • 直接抛出幂等异常,用以说明该请求为重复请求;
  • 直接返回上次请求一致的返回结果;

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

  • 如果是异常方案,客户端在调用幂等接口时需要对异常进行捕获,然后通过其他 API 获取上次请求的处理结果,根据结果不同来决定接下来的处理流程;
  • 如果是直接返回上次请求的处理结果,则客户端不需要做额外的处理,直接使用重试机制对请求进行重新发送即可,获取结果后自然进入到下面的流程;

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

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

3.5. 幂等组件

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

整体设计如下图所示:

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

整体设计比较简单,运行流程如下:

  • IdempotentInterceptor 会对 @Idempotent 注解标记的方法进行拦截;
  • 当方法第一次被调用时,会读取 @Idempotent 注解上的配置信息,使用 IdempotentExecutorFactoris 为每个方法创建一个 IdempotentExecutor 实例;
  • 在方法调用时,将请求直接路由到 IdempotentExecutor 实例,由 IdempotentExecutor 完成核心流程;
  • 其中,IdempotentExecutorFactories 拥有多个 IdempotentExecutorFactory 实例,并根据 @Idempotent 上配置的 executorFactory 属性使用对应的实例完成创建工作;

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

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

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

IdempotentExecutor 处理核心流程如下:

  • 通过 SpringEL 表达式从入参中提取 unique key 信息;
  • 根据 group 和 unique key 从 ExecutionRecordRepository 中读取执行记录 ExecutionRecord;
  • 如果 ExecutionRecord 为已完成状态,则根据配置直接返回 ExecutionRecord 的执行结果 或者 直接抛出 RepeatedSubmitException 异常;
  • 如果 ExecutionRecord 为执行中,则出现并发问题,直接抛出 ConcurrentRequestException 异常;
  • 如果 ExecutionRecord 为未执行,先执行方法获取返回值,然后使用 ExecutionRecordRepository 对 ExecutionRecord 进行更新,然后返回执行结果;

使用前需要引入 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 为核心配置,详细信息如下:

  • executorFactory 为 IdempotentExecutorFactory,及在 IdempotentConfiguration 中配置的bean,默认为 DEFAULT_EXECUTOR_FACTORY
  • group 为组信息,用于区分不同的业务场景,同一业务场景使用相同的配置;
  • keyEl 为提取幂等键所用的 SpringEl 表达式,#key 说明入参的 key 将作为幂等键,group + key 为一个完整的幂等键,唯一识别一次请求;
  • handleType 是处理类型,及重复提交时如何处理请求

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));
    }
}

运行测试用例,测试通过,可以得出:

  • 第一次操作,与正常方法一致,成功返回结果值;
  • 第二次操作,直接抛出 RepeatedSubmitException 异常,同时方法未执行,结果与第一次调用一致;
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));
    }
}

运行测试用例,用例通过,可知:

  • 第一次操作,与方法逻辑一致,更新数据并抛出 IdempotentTestException 异常;
  • 第二次操作,直接抛出 IdempotentTestException 异常,同时方法未执行,结果与第一次一致;
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());
}

运行单元测试,测试通过,核心测试逻辑如下:

  • 创建一个线程,执行耗时方法;
  • 等待 1 秒后,主线程也执行耗时方法;
  • 此时,两个线程并发执行耗时方法,后进入的主线程直接抛出 ConcurrentRequestException;
3.5.5. redis 支持

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

引入 redis 非常简单,大致分两步:

  • 在 IdempotentConfiguration 中注册 redisExecutorFactory bean;
  • @Idempotent 注解中使用 redisExecutorFactory 即可;

添加 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 机制:

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

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

  • 服务端幂等。幂等作为一种通用能力,建议与业务逻辑进行分离,在需要的时候直接在业务方法上进行配置即可,但仍旧有一些最佳实践:
  • 使用直接返回上次的处理结果替代异常中断,使调用方更加简洁;
  • 业务异常是一种特殊的返回值,也需要进行幂等保护;
  • 有幂等保护后,仍旧需要对并发请求进行处理,此时直接通过异常对重复流程进行中断即可;

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



Tags:DDD   点击:()  评论:()
声明:本站部分内容及图片来自互联网,转载是出于传递更多信息之目的,内容观点仅代表作者本人,不构成投资建议。投资者据此操作,风险自担。如有任何标注错误或版权侵犯请与我们联系,我们将及时更正、删除。
▌相关推荐
DDD 与 CQRS 才是黄金组合
在日常工作中,你是否也遇到过下面几种情况: 使用一个已有接口进行业务开发,上线后出现严重的性能问题,被老板当众质疑:“你为什么不使用缓存接口,这个接口全部走数据库,这怎么能扛...【详细内容】
2024-03-27  Search: DDD  点击:(11)  评论:(0)  加入收藏
DDD死党:单引擎查询利器
基于索引的单表查询,是 MySQL 正确打开方式!基于 QueryObject 的声明式查询,是简单查询的正确使用方式!1、应用场景单表查询在业务开发中占比最大,是所有 CRUD Boy 的入门必备,所...【详细内容】
2023-12-19  Search: DDD  点击:(121)  评论:(0)  加入收藏
DDD死党:内存Join——将复用和扩展用到极致
1. 为什么"内存Join"是个无法绕过的话题首先,我们先简单解释下,什么是“内存Join”。相信大家对关系数据库的 join 语句肯定不陌生,其作用就是通过关联关系从多个表中查询数据,...【详细内容】
2023-12-14  Search: DDD  点击:(211)  评论:(0)  加入收藏
我们聊聊DDD、SOA、微服务和微内核
DDD、SOA、微服务和微内核,看到经常有人把这几个概念拿出来一起讲。事实上,DDD和其他三个不是一个维度的东西。DDD其实特别好理解,DDD就是领域来驱动设计嘛,是一种设计思想。很...【详细内容】
2023-12-08  Search: DDD  点击:(231)  评论:(0)  加入收藏
DDD架构下的防御式编程:5大关卡共同保障业务数据的有效性
一般情况下,在流程达到存储引擎前,所有的验证规则必须全部通过,尽量不要使用存储引擎作为兜底方案。但有一种情况极为特殊,也就只有存储引擎能够优雅的完成,那就是唯一键保护。1....【详细内容】
2023-12-03  Search: DDD  点击:(140)  评论:(0)  加入收藏
DDD四层微服务架构
一、微服务搭建思路大家看到的这张架构图并不是空穴来潮,它是通过不断演变出来的,我们要从DDD四层架构、微服务架构两个维度去融合理解。这里的DDD四层架构适用于单个服务的工...【详细内容】
2023-11-24  Search: DDD  点击:(216)  评论:(0)  加入收藏
DDD 必备架构--六边形架构
架构是研究“分”和“合”的艺术,通过“分离关注点”将系统拆分为多个部分,然后在“原则和规则”的约束下对组件进行装配,形成高内聚的构件;再根据需求对多个构件进行关联,形成低...【详细内容】
2023-11-09  Search: DDD  点击:(374)  评论:(0)  加入收藏
DDD 与 CQRS 才是黄金组合,你觉得呢?
“数据密集型系统”越来越多的应用程序有着各种严格而广泛的要求,单个工具不足以满足所有的数据处理和存储需求。取而代之的是,总体工作被拆分成一系列能被单个工具高效完成的...【详细内容】
2023-11-08  Search: DDD  点击:(286)  评论:(0)  加入收藏
DDD与微服务集成的第一战役:客户端重试&服务端幂等
当一个接口从简单的内部调用升级为远程方法调用(RPC)会面临很多问题,比如: 本地事务失效。在内部调用时,多个方法通常在同一事务中执行,可以使用本地数据库事务来确保数据的一致性...【详细内容】
2023-10-30  Search: DDD  点击:(380)  评论:(0)  加入收藏
去哪儿网架构演进之路:微服务的尽头原来是DDD……
一、架构设计理念与技术1.架构演变路径图片 单体(又称巨石系统):所有业务融合于一体。在项目早期,公司一般会选择单体以降低运营等各方面成本。 服务化:随着业务飞速发展和流量增...【详细内容】
2023-10-11  Search: DDD  点击:(265)  评论:(0)  加入收藏
▌简易百科推荐
Redis 不再 “开源”,未来采用 SSPLv1 和 RSALv2 许可证
Redis 官方于21日宣布修改开源协议 &mdash;&mdash; 未来所有版本都将使用 “源代码可用” 的许可证 (source-available licenses)。具体来说,Redis 将不再遵循 BSD 3-Clause...【详细内容】
2024-03-27  dbaplus社群    Tags:Redis   点击:(11)  评论:(0)  加入收藏
Redis“叛逃”开源,得罪了几乎所有人
内存数据库供应商Redis近日在开源界砸下了一块“巨石”。Redis即将转向双许可模式,并实施更为严格的许可条款。官方对此次变更的公告直截了当:从Redis 7.4版本开始,Redis将在Re...【详细内容】
2024-03-25    51CTO  Tags:Redis   点击:(9)  评论:(0)  加入收藏
如何使用 Redis 实现消息队列
Redis不仅是一个强大的内存数据存储系统,它还可以用作一个高效的消息队列。消息队列是应用程序间或应用程序内部进行异步通信的一种方式,它允许数据生产者将消息放入队列中,然...【详细内容】
2024-03-22  后端Q  微信公众号  Tags:Redis   点击:(17)  评论:(0)  加入收藏
Redis不再 “开源”
Redis 官方今日宣布修改开源协议 &mdash;&mdash; 未来所有版本都将使用 “源代码可用” 的许可证 (source-available licenses)。具体来说,Redis 将不再遵循 BSD 3-Clause 开...【详细内容】
2024-03-21  OSC开源社区    Tags:Redis   点击:(8)  评论:(0)  加入收藏
在Redis中如何实现分布式锁的防死锁机制?
在Redis中实现分布式锁是一个常见的需求,可以通过使用Redlock算法来防止死锁。Redlock算法是一种基于多个独立Redis实例的分布式锁实现方案,它通过协调多个Redis实例之间的锁...【详细内容】
2024-02-20  编程技术汇    Tags:Redis   点击:(47)  评论:(0)  加入收藏
手动撸一个 Redis 分布式锁
大家好呀,我是楼仔。今天第一天开工,收拾心情,又要开始好好学习,好好工作了。对于使用 Java 的小伙伴,其实我们完全不用手动撸一个分布式锁,直接使用 Redisson 就行。但是因为这些...【详细内容】
2024-02-19  楼仔  微信公众号  Tags:Redis   点击:(39)  评论:(0)  加入收藏
工作中Redis有哪些好用的运维工具
工作中使用 Redis 时,如果大家公司没有专业运维,可能开发人员就会面临这些运维的工作,包括 Redis 的运行状态监控,数据迁移,主从集群、切片集群的部署和运维等等。本文我就从这三...【详细内容】
2024-02-06  waynaqua    Tags:Redis   点击:(55)  评论:(0)  加入收藏
批量执行Redis命令的四种方式!
前言在我们的印象中Redis命令好像都是一个个单条进行执行的,如果有人问你如何批量执行Redis命令,你能回答的上吗,或者说能答出几种方式呢?最容易想到的是Redis的一些批量命令,例...【详细内容】
2024-01-17  小许code  微信公众号  Tags:Redis命令   点击:(57)  评论:(0)  加入收藏
Redis 实现多规则限流的思考与实践
市面上很多介绍redis如何实现限流的,但是大部分都有一个缺点,就是只能实现单一的限流,比如1分钟访问1次或者60分钟访问10次这种,但是如果想一个接口两种规则都需要满足呢,我们的...【详细内容】
2024-01-03  架构精进之路  微信公众号  Tags:Redis   点击:(109)  评论:(0)  加入收藏
一站式Redis解决方案
Redis是一个高效的内存数据库,它支持包括String、List、Set、SortedSet和Hash等数据类型的存储,在Redis中通常根据数据的key查询其value值,Redis没有模糊条件查询,在面对一些需...【详细内容】
2024-01-01  大雷家吃饭    Tags:Redis   点击:(66)  评论:(0)  加入收藏
站内最新
站内热门
站内头条