幂等防重最佳实践_防重场景和防重方法

1、幂等定义

任意多次执行 和 执行一次 产生的业务影响相同

2、业务场景

3、Bad case

4、目标

调用方实现合理的重试策略，被调用方实现应对重试的幂等策略

• 判断出重复请求

单据号幂等、任务状态防重

• 处理重复请求

一般是拒绝、忽略、日志打出来

5、方案：

5.1 明确定义具体业务场景下什么是相同请求

5.2 判断重复方式

MQ、接口调用、数据库唯一键

唯一键约束-利用数据库的UniqueKey

• 最为常见的约束重复单据的方式，利用数据库的唯一键进行兜底

• 加了唯一索引之后，第一次请求数据可以插入成功。但后面的相同请求，插入数据时会报Duplicate entry 异常，表示唯一索引有冲突

• 业务中捕获DuplicateKeyException异常，并返回成功

• 和状态机update方式思想类似，不过状态机方式不用担心后续唯一键会改变 以及 也不用吃异常

唯一键约束-insert前先select

• 先查，没有再写逻辑

• 但是主从延时场景下，select可能不准 -> ，主库压力不大场景下，可以走主select

• 但是高并发的场景下，t1和t2，即使都select主，发现都没有数据，结果写时，则会有一个写报错或不符合预期。

  可以在select之前，再加一层redis锁setNx，专门用来防并发

唯一键约束-分布式锁[redission、zk]

但分布式锁只能保障一段时间内最多只有一个操作，无法保障无论什么时候都只允许一个操作（除非不设置锁超时的时间也不释放锁）

状态机约束

• 任务状态流转：已创建1—>已落表2 ->已计算3 -> 全部完成8
• eg：两个重复的MQ(taskCode = A)并发的进来了。二者都需要将任务状态从1update到2。update更新语句执行结果返回result = 1代表更新成功
• thread1更新时，db中任务A状态为1，t1将A从1更新为2，result = 1
• 同时thread2也进行更细操作，db中任务A状态为2了，将A从2更新为2，result = 0
• 可以通过result值1和0，判断哪个thread是第一个MQ。t2是重复的MQ，后续操作就可以终止了。即防重了，又防并发了

点击展开内容

int result = update();
if(result <= 0) { 
	return;//是重复的任务，可以忽略掉
}
1
2
3
4

5.3 对于重复的处理

5.4 重试

6、实践

6.1 订单支付

代码1

public String buildLockName(String orderId){  
    return distributedLockNamePrefix + "_" + orderId; 
}
1
2
3

代码2

public static final  int orderPayLockExpireTime = 60; //分布式锁过期时间，单位秒    
 
public void payOrderId(String orderId){   
  String lockName = buildLockName(orderId);        
  Lock lock = distributedLockManager.getReentrantLock(lockName, orderPayLockExpireTime);        
  try{            
    // 尝试获取锁失败            
    if(!lock.tryLock()){                
      log.warn("尝试获取分布式锁失败，订单已经在处理中");                
      throw new Exception("订单已在处理中，请勿重复支付");           
    }            
    // 修改数据库订单状态为"支付中"            
    updateOrderToPaying(orderId);           
    // 真正处理支付请求            
    doOrderPay(orderId);           
    // 修改数据库订单状态为"支付成功"           
    updateOrderToSuccess(orderId);        
  }catch (Exception e){            
    // 异常处理....            
    handleException(e);        
  }finally {          
    lock.unlock();      
  }   
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

代码3

//更新数据库中的订单状态为支付中
public void updateOrderToPaying(String orderId) throws Exception { 
  // 根据订单id将订单更新为支付中        
  int row = orderRepo.updateOrderStatus(orderId, StatusEnum.ORDER_READY_TO_PAY.getCode(),StatusEnum.ORDER_PAYING.getCode();                                                       
   if (row < 1) {      
     throw new Exception("订单状态错误，订单初始状态不是待支付");    
   }    
 }
1
2
3
4
5
6
7
8

代码4

//真正处理支付请求  
public OrderPayResponse doOrderPay(String orderId){     
  try{           
    OrderPayRequest orderPayRequest = buildOrderPayRequest(orderId);      
    OrderPayResponse orderPayResponse = payTService.pay(orderPayRequest);  
    if(orderPayResponse.getCode != ResponseStatus.SUCCESS){        
      log.warn("请求支付平台返回错误,{}",orderPayResponse);            
      throw new BusinessException(OrderPayResponseCodeEnum.FAIL.getCode(),"支付平台返回错误");   
    }           
    return orderPayResponse;       
  } catch (Exception e){         
    log.error("请求支付平台发生异常，orderId：{}",orderId,e);         
    // 判断哪些异常是可以让业务发起重试          
    if(e instanceof ExceptionCanRetry){            
      // 可以直接让业务发起重试的异常             
      throw new BusinessException(OrderPayResponseCodeEnum.RETRY.getCode(),"支付平台返回错误,可重试");            }          
    // 其余情况都不能直接发起重试，需要明确告诉业务          
    throw new BusinessException(OrderPayResponseCodeEnum.NO_RETRY.getCode(),"支付平台返回错误,不可重试");        }  
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

//修改订单状态为支付成功
public void updateOrderToSuccess(String orderId){      
  // 根据订单id将订单更新为支付中       
  int row = orderRepo.updateOrderStatus(orderId, StatusEnum.ORDER_PAYING.getCode(),                StatusEnum.ORDER_PAY_SUCCESS.getCode();       
   if (row < 1) {         
     throw new Exception("更新订单成功错误，订单初始状态不是支付中");      
   }   
                                        }
1
2
3
4
5
6
7
8

代码6

` //处理异常   
  public void handleException(Exception e) throws Exception {    
  // 将订单状态变为失败        
  updateOrderToFail(orderId);        
  // 可以直接进行重试的异常        
  if (e instanceof ExceptionCanRetry) {           
    // write Into retryLog            
    // .....            throw e;        
  }        
  // 异常无法进行重试，报警处理，需要人工进行介入查看        
  alaramService.pushNotify();       
  throw e;    
} `
1
2
3
4
5
6
7
8
9
10
11
12
13

为什么分布式锁获取失败需要直接丢弃，而不是阻塞等待

此处分布式锁的作用是为了防止单据重复，业务特性确认一个订单只能有一次请求，所以重复请求直接丢弃

为什么有了数据库状态的单向变更乐观锁，前面还需要缓存分布式锁去重操作？

减少极端情况下，大量重复请求对数据库的压力

​

给支付平台发送请求，为什么需要区分可重试和不可重试的两大类异常？

1. 针对特定的错误类型，系统做到可重试，无需人工介入

   2.针对不可重试错误类型，必须人工和下游确认后再次进行发起，避免造成金额损失

6.2 账户余额扣减加操作：针对账户A，需要往账户A内加100元。

代码1

//给账户余额加上amount金额
public void addMountToAccount(String accountId,Long amount){       
  String lockName = buildLockName(accountId);        
  Lock lock = distributedLockManager.getReentrantLock(lockName, accountLockExpireTime);        
  try{            
    lock.lock();           
    // 获取账户原始余额           
    Long originAmount = getAccountAmount(accountId);             
    // 更新后的余额           
    Long newAmount = originAmount + amount;            
    // 更新账户余额           
    updateAccountAmount(accountId,newAmount,originAmount);        
  }catch (Exception e){            
    // 异常处理....            
    handleException(e);       
  }finally {       
    lock.unlock();     
  }   
} `
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

代码2

/**     
* 更新账户余额     *    
* @param accountId    账户id    
* @param originAmount 库中现有余额   
* @param newAmount    更新后的余额     
* @return     */   
public void updateAccountAmount(String accountId, Long originAmount, Long newAmount) {  
  // update account set amount = $newAmount where id = $accountId and amount = $originAmount;        
  int row = accountRepo.updateAmountByAccountId(accountId, originAmount, newAmount);        
  if (row < 1) {            
    throw new Exception("更新账户余额错误");     
  }    
}
1
2
3
4
5
6
7
8
9
10
11
12
13

代码3

@Schedule("reprocess_task")//执行定时补偿任务
public void reProcessTask(){     
  Integer offset = 0;       
  Integer limit = 10;       
  while (true) {          
    List<Task> taskList =  reTryLogRepo.getUnProcessList(offset,limit);         
    // 未找到待执行的任务，则退出         
    if(CollectionUtils.isEmpty(taskList)){         
      return;          
    }                      
    for(Task task:taskList){       
      // 重试任务中的重试次数+1       
      task.addRetryTime();          
      boolean resultFlag = doReprocess(task);         
      if(resultFlag){                 
        // task置为成功                 
        task.setTaskSuccess();          
      }                               
      // 更新DB中的task信息，            
      //.....                              
      // 判断任务是否达到重试最大次数             
      if(task.reachMaxRetryTime()){                  
        // 报警                
      }           
    }           
    offset += limit;     
  }   
} 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

为什么未获取到锁需要阻塞等待

• 账户流水操作串行化，同一时间会有多个线程对账户余额进行加减操作
• 获取锁失败，需要等待其他线程完成对账户的操作后，继续执行，不能直接丢弃

为什么不能直接用数据库余额累加。amount+=addAmount

例：账户余额有300元，需要给账户加100元。由于某种原因，重复请求给了DB。

• 1.如果用amount+= addAmount。 update account set amount = amount+100 where id=1234 被执行两次
• 2.如果用悲观锁条件。 update account set amount=400 where id=1234 and amount = 300 被执行两次。

明显方案1中会出现金额错误

为什么获取余额时，不用强制读主库，主从延迟会不会有影响？

不会有影响，update操作的where条件会强制读主库。当主从延迟存在时候，此时更新会报错

为什么所有异常一律进入reTryLog，不用区分不同异常来做重试

• 因为针对账户余额操作，收到操作，表明最终必须成功
• 账户余额操作不涉及分布式事务，没有下游的事务错误

7、中间件使用规范