修改死锁重试机制
This commit is contained in:
+57
-9
@@ -576,15 +576,19 @@ public class CustomerAccountApiImpl implements ICustomerAccountApi
|
|||||||
private CustomerAccount getOrInitCnyAccount(Long customerId, String username)
|
private CustomerAccount getOrInitCnyAccount(Long customerId, String username)
|
||||||
{
|
{
|
||||||
CustomerAccount account = customerMoneyMapper.selectAccountForUpdate(customerId, BASE_CURRENCY);
|
CustomerAccount account = customerMoneyMapper.selectAccountForUpdate(customerId, BASE_CURRENCY);
|
||||||
if (account == null)
|
if (account != null)
|
||||||
|
{
|
||||||
|
return account;
|
||||||
|
}
|
||||||
|
BmsCustomer customer = bmsCustomerMapper.selectRawCreditByCustomerId(customerId);
|
||||||
|
if (customer == null)
|
||||||
|
{
|
||||||
|
throw new ServiceException("客户不存在");
|
||||||
|
}
|
||||||
|
BigDecimal initCredit = nz(customer.getCreditEd());
|
||||||
|
BigDecimal initTemp = nz(customer.getTempEd());
|
||||||
|
try
|
||||||
{
|
{
|
||||||
BmsCustomer customer = bmsCustomerMapper.selectRawCreditByCustomerId(customerId);
|
|
||||||
if (customer == null)
|
|
||||||
{
|
|
||||||
throw new ServiceException("客户不存在");
|
|
||||||
}
|
|
||||||
BigDecimal initCredit = nz(customer.getCreditEd());
|
|
||||||
BigDecimal initTemp = nz(customer.getTempEd());
|
|
||||||
account = createCnyAccount(customerId, initCredit, initTemp, username);
|
account = createCnyAccount(customerId, initCredit, initTemp, username);
|
||||||
writeInitFlow(account, CustomerCreditFlow.CREDIT_TYPE_CREDIT,
|
writeInitFlow(account, CustomerCreditFlow.CREDIT_TYPE_CREDIT,
|
||||||
initCredit, null, null, username, "首次创建CNY账户,从客户档案同步初始额度");
|
initCredit, null, null, username, "首次创建CNY账户,从客户档案同步初始额度");
|
||||||
@@ -593,8 +597,52 @@ public class CustomerAccountApiImpl implements ICustomerAccountApi
|
|||||||
writeInitFlow(account, CustomerCreditFlow.CREDIT_TYPE_TEMP,
|
writeInitFlow(account, CustomerCreditFlow.CREDIT_TYPE_TEMP,
|
||||||
initTemp, null, null, username, "首次创建CNY账户,从客户档案同步临时额度");
|
initTemp, null, null, username, "首次创建CNY账户,从客户档案同步临时额度");
|
||||||
}
|
}
|
||||||
|
return account;
|
||||||
}
|
}
|
||||||
return account;
|
catch (RuntimeException e)
|
||||||
|
{
|
||||||
|
// Concurrent first-create: peer may have inserted; re-lock existing row.
|
||||||
|
if (isAccountInitRace(e))
|
||||||
|
{
|
||||||
|
account = customerMoneyMapper.selectAccountForUpdate(customerId, BASE_CURRENCY);
|
||||||
|
if (account != null)
|
||||||
|
{
|
||||||
|
log.warn("CNY account created by peer, reuse customerId={}", customerId);
|
||||||
|
return account;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
throw e;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Deadlock / duplicate key while inserting customer_account. */
|
||||||
|
private static boolean isAccountInitRace(Throwable e)
|
||||||
|
{
|
||||||
|
Throwable t = e;
|
||||||
|
while (t != null)
|
||||||
|
{
|
||||||
|
String name = t.getClass().getName();
|
||||||
|
if (name.indexOf("DeadlockLoser") >= 0
|
||||||
|
|| name.indexOf("DuplicateKey") >= 0
|
||||||
|
|| name.indexOf("MySQLTransactionRollback") >= 0
|
||||||
|
|| name.indexOf("DataIntegrityViolation") >= 0)
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
String msg = t.getMessage();
|
||||||
|
if (msg != null)
|
||||||
|
{
|
||||||
|
String m = msg.toLowerCase();
|
||||||
|
if (m.indexOf("deadlock") >= 0
|
||||||
|
|| m.indexOf("duplicate entry") >= 0
|
||||||
|
|| m.indexOf("try restarting transaction") >= 0)
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
t = t.getCause();
|
||||||
|
}
|
||||||
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** 创建CNY账户(余额0,已用0) */
|
/** 创建CNY账户(余额0,已用0) */
|
||||||
|
|||||||
+72
-1
@@ -96,6 +96,9 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
|
|||||||
return task;
|
return task;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static final int BILL_DB_RETRY_MAX = 3;
|
||||||
|
private static final long BILL_DB_RETRY_BASE_MS = 80L;
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void processTask(Long taskId)
|
public void processTask(Long taskId)
|
||||||
{
|
{
|
||||||
@@ -167,7 +170,7 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
|
|||||||
log.info("billing task REFUND path uses bill upsert taskId={}", taskId);
|
log.info("billing task REFUND path uses bill upsert taskId={}", taskId);
|
||||||
}
|
}
|
||||||
|
|
||||||
TmsPriceQuoteResult result = prcOrderBillService.bill(billReq);
|
TmsPriceQuoteResult result = billWithDbRetry(taskId, billReq);
|
||||||
writeback(task, BillTaskSuccessStatus.SUCCESS, result == null ? null : result.getBillId(),
|
writeback(task, BillTaskSuccessStatus.SUCCESS, result == null ? null : result.getBillId(),
|
||||||
null);
|
null);
|
||||||
}
|
}
|
||||||
@@ -178,6 +181,74 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Retry bill() on MySQL deadlock / lock wait (common when many tasks share one customer_account).
|
||||||
|
*/
|
||||||
|
private TmsPriceQuoteResult billWithDbRetry(Long taskId, TmsPriceBillRequest billReq)
|
||||||
|
{
|
||||||
|
RuntimeException last = null;
|
||||||
|
for (int attempt = 1; attempt <= BILL_DB_RETRY_MAX; attempt++)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return prcOrderBillService.bill(billReq);
|
||||||
|
}
|
||||||
|
catch (RuntimeException e)
|
||||||
|
{
|
||||||
|
last = e;
|
||||||
|
if (attempt < BILL_DB_RETRY_MAX && isRetryableDbConflict(e))
|
||||||
|
{
|
||||||
|
long sleepMs = BILL_DB_RETRY_BASE_MS * attempt
|
||||||
|
+ (long) (Math.random() * BILL_DB_RETRY_BASE_MS);
|
||||||
|
log.warn("billing task db conflict, retry taskId={} attempt={}/{} sleepMs={} msg={}",
|
||||||
|
taskId, attempt, BILL_DB_RETRY_MAX, sleepMs, cut(e.getMessage(), 200));
|
||||||
|
try
|
||||||
|
{
|
||||||
|
Thread.sleep(sleepMs);
|
||||||
|
}
|
||||||
|
catch (InterruptedException ie)
|
||||||
|
{
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
throw e;
|
||||||
|
}
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
throw e;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
throw last != null ? last : new ServiceException("billing retry exhausted");
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Deadlock / lock wait / duplicate account insert — safe to retry whole bill txn. */
|
||||||
|
static boolean isRetryableDbConflict(Throwable e)
|
||||||
|
{
|
||||||
|
Throwable t = e;
|
||||||
|
while (t != null)
|
||||||
|
{
|
||||||
|
String name = t.getClass().getName();
|
||||||
|
if (name.indexOf("DeadlockLoser") >= 0
|
||||||
|
|| name.indexOf("CannotAcquireLock") >= 0
|
||||||
|
|| name.indexOf("MySQLTransactionRollback") >= 0)
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
String msg = t.getMessage();
|
||||||
|
if (msg != null)
|
||||||
|
{
|
||||||
|
String m = msg.toLowerCase();
|
||||||
|
if (m.indexOf("deadlock") >= 0
|
||||||
|
|| m.indexOf("try restarting transaction") >= 0
|
||||||
|
|| m.indexOf("lock wait timeout") >= 0
|
||||||
|
|| m.indexOf("duplicate entry") >= 0)
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
t = t.getCause();
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public PrcBillingTask getById(Long taskId)
|
public PrcBillingTask getById(Long taskId)
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user