真实项目中使用悲观锁或者乐观锁来解决并发问题

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

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 id="updateOptimisticByPrimaryKey" parameterType="com.ryx.xiaoxin_distribute.entity.pojo.RyxAccount">
        UPDATE `ryx_account`
        <set>
            <if test="bean.name != null">
                `name` = #{bean.name},
            </if>
            <if test="bean.money != null">
                `money` = #{bean.money},
            </if>
            <if test="bean.createTime != null">
                `createTime` = #{bean.createTime},
            </if>
            <if test="bean.updateTime != null">
                `updateTime` = #{bean.updateTime},
            </if>
            version = version +1
        </set>
        where `id`=#{id}
        and version = #{bean.version}
    </update>

主要执行的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;
    }

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