本人在金融公司任職,今天來分享下關于轉賬的一些并發處理問題,這節內容,我們不聊實現原來,就單純的看看如何實現
廢話不多說,咱們直接開始,首先我會模擬一張轉賬表
如下圖所示:

image.png

一張簡單的賬戶表,有name,賬戶余額等等,接下來我將用三種鎖的方式來實現下并發下的互相轉賬
一:悲觀鎖:
概念我就在這里不說了,很簡單,直接上代碼
接口層:

void transfer(Integer sourceId, Integer targetId, BigDecimal money);

實現層:

/**
* 轉賬操作(使用悲觀鎖的方式執行轉賬操作)
* source:轉出賬戶
* target:轉入專戶
* money:轉賬金額
* @param sourceId
* @param targetId
*/
@Override
@Transactional(rollbackFor = Exception.class)
public void transfer(Integer sourceId, Integer targetId, BigDecimal money) {
RyxAccount source;
RyxAccount target;
//處理死鎖問題,每次從小到大執行
if (sourceId <= targetId){
source = getAccount(sourceId);
target = getAccount(targetId);
}else{
target = getAccount(targetId);
source = getAccount(sourceId);
}

if (source.getMoney().compareTo(money) >=0){
source.setMoney(source.getMoney().subtract(money));
target.setMoney(target.getMoney().add(money));

updateAccount(source);
updateAccount(target);
}else{
log.error("賬戶[{}]余額[{}]不足,不允許轉賬", source.getId(),source.getMoney());
}

}

private RyxAccount getAccount(Integer sourceId) {
return this.ryxAccountService.getRyxAccountByPrimaryKeyForUpdate(sourceId);
}

private void updateAccount(RyxAccount account) {
account.setUpdateTime(new Date());
this.ryxAccountService.updateByPrimaryKey(account, account.getId());
}

mapper層:

<select id="getRyxAccountByPrimaryKeyForUpdate" resultMap="base_result_map" >
select <include refid="base_column_list" /> from `ryx_account` where `id`=#{id} for update
</select>

測試代碼:我同時啟動5個線程,來執行轉賬操作

@Test
public void transferTest() throws InterruptedException {
Integer zhangsanAccountId = 315;
Integer lisiAccountId = 316;
Integer wangwuAccountId =317;
Integer zhaoliuAccountId = 318;

BigDecimal money = new BigDecimal(100);
BigDecimal money1 = new BigDecimal(50);
//zhangsan轉lisi100
Thread t1 = new Thread(() ->{
accountService.transfer(zhangsanAccountId, lisiAccountId, money);
});

//lisi轉wangwu100
Thread t2 = new Thread(() ->{
accountService.transfer(lisiAccountId, wangwuAccountId, money);
});

//wangwu轉zhaoliu 100
Thread t3 = new Thread(() ->{
accountService.transfer(wangwuAccountId, zhaoliuAccountId, money);
});

//zhoaliu轉zhangsan 100
Thread t4 = new Thread(() ->{
accountService.transfer(zhaoliuAccountId, zhangsanAccountId, money);
});

//zhangsan轉zhaoliu 50
Thread t5 = new Thread(() ->{
accountService.transfer(zhangsanAccountId, zhaoliuAccountId, money1);
});


t1.start();
t2.start();
t3.start();
t4.start();
t5.start();

t1.join();
t2.join();
t3.join();
t4.join();
t5.join();
}

啟動我們來看看結果

image.png

執行結果符合預期
悲觀鎖的主要邏輯就是在查詢的時候使用select * ..... for update 語句,還有一點,子啊賬戶查詢的時候,我們按照主鍵的順序執行了排序后進行處理,否則會發生死鎖,原理我們就先不介紹了,主要就是先看看如何處理

二:悲觀鎖:這個情況下,我就就需要在數據庫中增加一個版本號,

image.png

接著看代碼
接口層:

/**
* 樂觀鎖方式
* @param sourceId 轉出賬戶
* @param targetId 轉入賬戶
* @param money 轉賬金額
*/
void transferOptimistic(Integer sourceId, Integer targetId, BigDecimal money);

實現層:

/**
* 使用樂觀鎖執行
* @param sourceId
* @param targetId
* @param money
*/
@Override
@Transactional(rollbackFor = Exception.class)
public void transferOptimistic(Integer sourceId, Integer targetId, BigDecimal money) {
RyxAccount source = getAccountOptimistic(sourceId);
RyxAccount target = getAccountOptimistic(targetId);


if (source.getMoney().compareTo(money) >=0){
source.setMoney(source.getMoney().subtract(money));
target.setMoney(target.getMoney().add(money));

// 先鎖 id 較大的那行，避免死鎖
int result1, result2;
if (source.getId() <=target.getId()){
result1 = updateOptimisticAccount(source,source.getVersion());
result2 = updateOptimisticAccount(target,target.getVersion());
}else{
result2 = updateOptimisticAccount(target,target.getVersion());
result1 = updateOptimisticAccount(source,source.getVersion());
}

if (result1 < 1 || result2 < 1) {
throw new RuntimeException("轉賬失敗,重試中....");
} else {
log.info("轉賬成功");
}
}else{
log.error("賬戶[{}]余額[{}]不足,不允許轉賬", source.getId(),source.getMoney());
}

}
private int updateOptimisticAccount(RyxAccount account,Integer version) {
account.setUpdateTime(new Date());
return this.ryxAccountService.updateOptimisticByPrimaryKey(account, account.getId(),version);
}

mapper層:

UPDATE `ryx_account`
`name` = #{bean.name},
`money` = #{bean.money},
`createTime` = #{bean.createTime},
`updateTime` = #{bean.updateTime},
version = version +1
where `id`=#{id}
and version = #{bean.version}

主要執行的sql語句就是update set xxxx version = version+1 where version = #{bean.version}
我們來看看執行結果,測試代碼就不展示了,和上面一樣

image.png

可以看到報錯了,需要重試,所以如果你需要使用樂觀鎖的話需要,有重試機制,而且重試次數比較多,所以對于轉賬操作,就不適合使用
樂觀鎖去解決
三:分布式鎖:
接下來,我們在用第三種方式處理一下,就是使用分布式鎖,分布式鎖可以用redis分布式鎖,也可以使用zk做分布式鎖,比較簡單
我們就直接上代碼吧
接口層:

/**
* 分布式鎖方式
* @param sourceId 轉出賬戶
* @param targetId 轉入賬戶
* @param money 轉賬金額
*/
void transferDistributed(Integer sourceId, Integer targetId, BigDecimal money);

實現層:

/**
* 使用分布式鎖執行
* @param sourceId
* @param targetId
* @param money
*/
@Override
@Transactional(rollbackFor = Exception.class)
public void transferDistributed(Integer sourceId, Integer targetId, BigDecimal money) {
try {
accountLock.lock();
distributedAccount(sourceId,targetId,money);
} catch (Exception e) {
log.error(e.getMessage());
//throw new RuntimeException("錯誤啦");
} finally {
accountLock.unlock();
}
}

private void distributedAccount(Integer sourceId, Integer targetId, BigDecimal money) {
RyxAccount source;
RyxAccount target;

//解決死鎖問題
if (sourceId <= targetId){
source = this.ryxAccountService.getRyxAccountByPrimaryKey(sourceId);
target = this.ryxAccountService.getRyxAccountByPrimaryKey(targetId);
}else{
target = this.ryxAccountService.getRyxAccountByPrimaryKey(targetId);
source = this.ryxAccountService.getRyxAccountByPrimaryKey(sourceId);
}

if (source.getMoney().compareTo(money) >=0){
source.setMoney(source.getMoney().subtract(money));
target.setMoney(target.getMoney().add(money));

updateAccount(source);
updateAccount(target);
}else{
log.error("賬戶[{}]向[{}]轉賬余額[{}]不足,不允許轉賬", source.getId(),target.getId(),source.getMoney());
throw new RuntimeException("賬戶余額不足,不允許轉賬");
}
}

@Override
public void afterPropertiesSet() throws Exception {
if (accountLock == null){
accountLock = this.redisLockRegistry.obtain("account-lock");
}
}

分布式鎖的配置,可以參考我之前的筆記,這里在簡單貼出代碼

@Configuration
public class RedisLockConfiguration {

@Bean
public RedisLockRegistry redisLockRegistry(RedisConnectionFactory redisConnectionFactory) {
RedisLockRegistry redisLockRegistry = new RedisLockRegistry(redisConnectionFactory, "spring-cloud", 5000L);
return redisLockRegistry;
}

}
``
寫下來還是比較簡單,今天只是大概在代碼實現方面做了介紹,`咩有涉及到原理,原理分析篇涉及到的內容比較多,等后續整理出來再分享
再說說我們公司用的方法,我鎖在職的是金融公司,轉賬操作特別頻發,而且每天幾十個億的資金也很正常
而我們公司在處理轉賬的邏輯中,涉及到并發的時候,就是使用的悲觀鎖的方式來處理的.
今天就分享到這里!

總結

以上是生活随笔為你收集整理的悲观锁和乐观锁_悲观锁和乐观锁处理并发操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。