百度360必应搜狗淘宝本站头条
nodejsbufferorg.jsoupsetnocountoncv2.bitwise_nottime.strptime
当前位置：网站首页 > 编程字典 > 正文

“幂等”不等于“分布式锁”，也不得不考虑返回值

toyiye 2024-06-21 19:08 18 浏览 0 评论

1. 概览

在分布式系统中，幂等是一个非常重要的概念，常常与“重试”一起出现。当调用一个远程服务发生超时，调用方并不知道请求是否执行成功，这就是典型的“第三态”问题。对于这个问题最常见的解决方案便是进行主动重试，假如该操作是一个数据库插入操作，重试将对系统产生副作用（创建多条记录），这时我们常常会说，被调用接口需要保障幂等。

1.1. 背景

幂等可以简单定义如下：任意多次执行所产生的影响均与第一次执行的影响相同。

【注】从幂等定义上看，重心放在了操作之后的影响，及多次操作不会破坏内部状态。但在实际工作当中，除了内部状态外，接口的返回值也是一个重要要素，多次重复操作返回相同的结果往往更符合使用者的预期。

举个例子，在订单系统中，用户使用优惠券下单后会调用冻结接口对优惠券进行冻结操作，站着订单系统的视角，当进行冻结重试时你期望：

  1. 优惠券抛出异常，告知订单优惠券已经冻结；
  2. 优惠券直接返回上次的结果 “冻结成功；

大家可以思考下两种方案在下游使用时的异同，当然最好提供两种机制，由使用方根据场景进行定制。

在不同的场景下，幂等保护的方案是不同的，常见幂等处理策略有：

  1. 天然具有幂等性，不需要保护。比如读操作、按主键数据删除等；
  2. 使用上游信息作为下游主键或唯一键的插入场景，由于有键的约束，可以保障幂等。比如以 userId 作为 Card（名片）的主键；
  3. 数据库字段的直接更新操作也具有一定的幂等性。比如用户修改姓名；
  4. 直接使用外部存储，以幂等键为标识，对方法执行情况进行记录，并以此为判断依据完成幂等保护；

由于其他策略与场景强绑定，idempotent 重心放在方案4上，已覆盖更多的业务场景。

1.2. 目标

快速为非幂等接口增加幂等保护。

  1. 基于“能力声明化”的方式，为接口快速添加幂等保护；
  2. 支持常见的两种幂等保护策略；
    1. 直接返回上次的执行结果；
    2. 抛出异常告知重复提交；
  3. 支持存储层的扩展，并提供常见的存储实现，包括：
    1. redis 实现，适用于一般通用场景，性能好但缓存过期后可能存在幂等不生效的情况；
    2. db 实现，适用于比较严格的场景，比如与订单、金额相关的业务；

2. 快速入门

2.1. 准备工作

首先，引入 lego starter，在 maven pom 中添加如下信息：

<groupId>com.geekhalo.lego</groupId>
<artifactId>lego-starter</artifactId>
<version>0.1.15-idempotent-SNAPSHOT</version>

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

@Configuration
public class IdempotentConfiguration extends IdempotentConfigurationSupport {
    @Bean("dbExecutorFactory")
    public IdempotentExecutorFactory redisExecutorFactory(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 配置，可以自行到网上进行检索。

2.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 为核心配置，详细信息如下：

  1. executorFactory 为 IdempotentExecutorFactory，及在 IdempotentConfiguration 中配置的bean，默认为 DEFAULT_EXECUTOR_FACTORY
  2. group 为组信息，用于区分不同的业务场景，同一业务场景使用相同的配置；
  3. keyEl 为提取幂等键所用的 SpringEl 表达式，#key 说明入参的 key 将作为幂等键，group + key 为一个完整的幂等键，唯一识别一次请求；
  4. handleType 是处理类型，及重复提交时如何处理请求
    1. RESULT，直接返回上次的执行结果
    2. 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));
    }
}

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

  1. 第一次操作，与正常方法一致，成功返回结果值；
  2. 第二次操作，逻辑方法未执行，直接返回第一次的运行结果；

这是最常见的一种工作模式，除直接返回上次执行结果外，当发生重复提交时也可以抛出异常中断流程，只需将 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));
    }
}

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

  1. 第一次操作，与正常方法一致，成功返回结果值；
  2. 第二次操作，直接抛出 RepeatedSubmitException 异常，同时方法未执行，结果与第一次调用一致；

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

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

  1. 第一次操作，与方法逻辑一致，更新数据并抛出 IdempotentTestException 异常；
  2. 第二次操作，直接抛出 IdempotentTestException 异常，同时方法未执行，结果与第一次一致；

2.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. 创建一个线程，执行耗时方法；
  2. 等待 1 秒后，主线程也执行耗时方法；
  3. 此时，两个线程并发执行耗时方法，后进入的主线程直接抛出 ConcurrentRequestException；

2.5. Redis 支持

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

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

  1. 在 IdempotentConfiguration 中注册 redisExecutorFactory bean；
  2. @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 机制造成的重复请求。

3. 设计&扩展

3.1. 整体结构

image

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

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

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

3.2. 核心流程

image

IdempotentExecutor处理核心流程如下：

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

4. 项目信息

项目仓库地址：https://gitee.com/litao851025/lego

项目文档地址：https://gitee.com/litao851025/lego/wikis/support/idempotent

相关推荐

为何越来越多的编程语言使用JSON（为什么编程）

JSON是JavascriptObjectNotation的缩写，意思是Javascript对象表示法，是一种易于人类阅读和对编程友好的文本数据传递方法，是JavaScript语言规范定义的一个子...

何时在数据库中使用 JSON（数据库用json格式存储）

在本文中，您将了解何时应考虑将JSON数据类型添加到表中以及何时应避免使用它们。每天?分享?最新?软件?开发?，Devops，敏捷?，测试?以及?项目?管理?最新?，最热门?的?文章?，每天?花?...

MySQL 从零开始:05 数据类型（mysql数据类型有哪些,并举例）

前面的讲解中已经接触到了表的创建，表的创建是对字段的声明，比如：上述语句声明了字段的名称、类型、所占空间、默认值和是否可以为空等信息。其中的int、varchar、char和decimal都...

JSON对象花样进阶（json格式对象）

一、引言在现代Web开发中，JSON（JavaScriptObjectNotation）已经成为数据交换的标准格式。无论是从前端向后端发送数据，还是从后端接收数据，JSON都是不可或缺的一部分。...

深入理解 JSON 和 Form-data（json和formdata提交区别）

在讨论现代网络开发与API设计的语境下，理解客户端和服务器间如何有效且可靠地交换数据变得尤为关键。这里，特别值得关注的是两种主流数据格式：...

JSON 语法（json 语法 priority）

JSON语法是JavaScript语法的子集。JSON语法规则JSON语法是JavaScript对象表示法语法的子集。数据在名称/值对中数据由逗号分隔花括号保存对象方括号保存数组JS...

JSON语法详解（json的语法规则）

JSON语法规则JSON语法是JavaScript对象表示法语法的子集。数据在名称/值对中数据由逗号分隔大括号保存对象中括号保存数组注意：json的key是字符串,且必须是双引号，不能是单引号...

MySQL JSON数据类型操作（mysql的json）

概述mysql自5.7.8版本开始，就支持了json结构的数据存储和查询，这表明了mysql也在不断的学习和增加nosql数据库的有点。但mysql毕竟是关系型数据库，在处理json这种非结构化的数据...

JSON的数据模式（json数据格式示例）

像XML模式一样，JSON数据格式也有Schema，这是一个基于JSON格式的规范。JSON模式也以JSON格式编写。它用于验证JSON数据。JSON模式示例以下代码显示了基本的JSON模式。{"...

前端学习——JSON格式详解（后端json格式）

JSON(JavaScriptObjectNotation)是一种轻量级的数据交换格式。易于人阅读和编写。同时也易于机器解析和生成。它基于JavaScriptProgrammingLa...

什么是 JSON:详解 JSON 及其优势（什么叫json）

现在程序员还有谁不知道JSON吗？无论对于前端还是后端，JSON都是一种常见的数据格式。那么JSON到底是什么呢？JSON的定义...

PostgreSQL JSON 类型:处理结构化数据

PostgreSQL提供JSON类型，以存储结构化数据。JSON是一种开放的数据格式，可用于存储各种类型的值。什么是JSON类型？JSON类型表示JSON(JavaScriptO...

JavaScript:JSON、三种包装类（javascript 包）

JOSN:我们希望可以将一个对象在不同的语言中进行传递，以达到通信的目的，最佳方式就是将一个对象转换为字符串的形式JSON（JavaScriptObjectNotation）-JS的对象表示法...

Python数据分析 只要1分钟 教你玩转JSON 全程干货

Json简介：Json，全名JavaScriptObjectNotation,JSON(JavaScriptObjectNotation(记号、标记))是一种轻量级的数据交换格式。它基于J...

比较一下JSON与XML两种数据格式?（json和xml哪个好）

JSON（JavaScriptObjectNotation）和XML（eXtensibleMarkupLanguage）是在日常开发中比较常用的两种数据格式，它们主要的作用就是用来进行数据的传...

取消回复欢迎 你 发表评论:

请填写验证码
一周热门
最近发表
标签列表