本人在金融公司任职,今天来分享下关于转账的一些并发处理问题,这节内容,我们不聊实现原来,就单纯的看看如何实现
废话不多说,咱们直接开始,首先我会模拟一张转账表
如下图所示:

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个线程,来执行转账操作

@Testpublic 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转lisi100Thread t1 = new Thread(() ->{            accountService.transfer(zhangsanAccountId, lisiAccountId, money);});

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

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

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

//zhangsan转zhaoliu 50Thread 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("账户余额不足,不允许转账");}}

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

分布式锁的配置,可以参考我之前的笔记,这里在简单贴出代码

@Configurationpublic class RedisLockConfiguration {

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

}``写下来还是比较简单,今天只是大概在代码实现方面做了介绍,`咩有涉及到原理,原理分析篇涉及到的内容比较多,等后续整理出来再分享再说说我们公司用的方法,我锁在职的是金融公司,转账操作特别频发,而且每天几十个亿的资金也很正常而我们公司在处理转账的逻辑中,涉及到并发的时候,就是使用的悲观锁的方式来处理的.今天就分享到这里!