当前位置： 首页 > article >正文

使用单体锁和分布式锁解决超卖问题

article 2024/11/14 9:13:24

使用单体锁和分布式锁解决超卖问题

超卖现象演示


@SpringBootTest
@RunWith(SpringRunner.class)
public class DistributeLockApplicationTests {

    @Autowired
    private OrderService orderService;


    /**
     * 模拟超卖
     */
    @Test
    public void concurrentOrder() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(5);

        CyclicBarrier cyclicBarrier = new CyclicBarrier(5);

        ExecutorService executorService = Executors.newFixedThreadPool(5);

        for (int i = 0; i < 5; i++) {
            executorService.execute(() -> {
                try {
                    //等待5个线程执行到await时，再去执行后面得到步骤； 让所有的线程在等待等到某一刻再去执行，实现并发的效果
                    cyclicBarrier.await();

                    Integer orderId = orderService.createOrder();
                    System.out.println("订单id: " + orderId + "  " + Thread.currentThread().getName());

                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    countDownLatch.countDown();
                }
            });

            //等到5个线程执行完毕后，再去结束主线程，不这样做数据库连接会被关闭
            countDownLatch.await();

            //关闭线程池
            executorService.shutdown();

        }


    }
}

创建订单相关数据的方法如下：

    @Transactional(rollbackFor = Exception.class)
    public Integer createOrder() throws Exception {
        Product product = productMapper.selectByPrimaryKey(purchaseProductId);
        if (product == null) {
            throw new Exception("购买商品：" + purchaseProductId + "不存在");
        }

        //商品当前库存
        Integer currentCount = product.getCount();

        //校验库存
        if (purchaseProductNum > currentCount) {
            throw new Exception("商品" + purchaseProductId + "仅剩" + currentCount + "件，无法购买");
        }

        //计算剩余库存
        int leftCount = currentCount - purchaseProductNum;
        //更新库存
        product.setCount(leftCount);

        // 将提交的数据添加到数据库中.
        product.setUpdateTime(new Date());
        product.setUpdateUser("xxx");
        productMapper.updateByPrimaryKeySelective(product);

        //创建订单
        Order order = new Order();
        order.setOrderAmount(product.getPrice().multiply(new BigDecimal(purchaseProductNum)));
        order.setOrderStatus(1);
        order.setReceiverName("xxx");
        order.setReceiverMobile("13311112222");
        order.setCreateTime(new Date());
        order.setCreateUser("xxx");
        order.setUpdateTime(new Date());
        order.setUpdateUser("xxx");
        orderMapper.insertSelective(order);

        //创建订单中的
        OrderItem orderItem = new OrderItem();
        orderItem.setOrderId(order.getId());
        orderItem.setProductId(product.getId());
        orderItem.setPurchasePrice(product.getPrice());
        orderItem.setPurchaseNum(purchaseProductNum);
        orderItem.setCreateUser("xxx");
        orderItem.setCreateTime(new Date());
        orderItem.setUpdateTime(new Date());
        orderItem.setUpdateUser("xxx");
        orderItemMapper.insertSelective(orderItem);
        return order.getId();
    }

单体锁

数据库update行锁

数据库在更新一行数据时，会自动对这一行数据加锁，借助这个特性，我们可以解决并发问题。
将原来在程序中更新库存改成直接在数据库中更新库存。传入

    /**
     * 使用数据库update行锁处理
     * 数据库在更新某一行记录的时候会这一行加锁，一个线程处理完了，下一个线程才能继续处理
     */
    @Transactional(rollbackFor = Exception.class)
    public Integer createOrderByUpdateLock() throws Exception {
        Product product = productMapper.selectByPrimaryKey(purchaseProductId);
        if (product == null) {
            throw new Exception("购买商品：" + purchaseProductId + "不存在");
        }

        //商品当前库存
        Integer currentCount = product.getCount();

        //校验库存
        if (purchaseProductNum > currentCount) {
            throw new Exception("商品" + purchaseProductId + "仅剩" + currentCount + "件，无法购买");
        }

//        //计算剩余库存
//        int leftCount = currentCount - purchaseProductNum;
//        //更新库存
//        product.setCount(leftCount);
//
//        // 将提交的数据添加到数据库中.
//        product.setUpdateTime(new Date());
//        product.setUpdateUser("xxx");
//        productMapper.updateByPrimaryKeySelective(product);

        //直接写入剩余库存，而不是在内存中计算出剩余库存后再写入数据库
        productMapper.updateProductCount(purchaseProductNum, "xxx", new Date(), product.getId());

        //创建订单
        Order order = new Order();
        order.setOrderAmount(product.getPrice().multiply(new BigDecimal(purchaseProductNum)));
        order.setOrderStatus(1);//待处理
        order.setReceiverName("xxx");
        order.setReceiverMobile("13311112222");
        order.setCreateTime(new Date());
        order.setCreateUser("xxx");
        order.setUpdateTime(new Date());
        order.setUpdateUser("xxx");
        orderMapper.insertSelective(order);

        //创建订单中的
        OrderItem orderItem = new OrderItem();
        orderItem.setOrderId(order.getId());
        orderItem.setProductId(product.getId());
        orderItem.setPurchasePrice(product.getPrice());
        orderItem.setPurchaseNum(purchaseProductNum);
        orderItem.setCreateUser("xxx");
        orderItem.setCreateTime(new Date());
        orderItem.setUpdateTime(new Date());
        orderItem.setUpdateUser("xxx");
        orderItemMapper.insertSelective(orderItem);
        return order.getId();
    }

通过synchronzied关键字

将校验库存以及扣减库存的代码放在synchroinzed代码块中；但是需要自己手动管理事务。如果使用事务注解，第一个线程进入执行完后，事务未提交，第二个线程执行时获取到的库存还是1，这样还是会有并发问题。

    /**
     * 通过synchronized关键字
     */
    public Integer createOrderBySynchronized1() throws Exception {
        Product product = null;
        synchronized (OrderService.class) {
            product = productMapper.selectByPrimaryKey(purchaseProductId);
            if (product == null) {
                throw new Exception("购买商品：" + purchaseProductId + "不存在 " + Thread.currentThread().getName());
            }

            //商品当前库存
            Integer currentCount = product.getCount();

            //校验库存
            if (purchaseProductNum > currentCount) {
                throw new Exception("商品" + purchaseProductId + "仅剩" + currentCount + "件，无法购买 " + Thread.currentThread().getName());
            }
            //计算剩余库存
            int leftCount = currentCount - purchaseProductNum;
            //更新库存
            product.setCount(leftCount);

            // 将提交的数据添加到数据库中.
            product.setUpdateTime(new Date());
            product.setUpdateUser("xxx");
            productMapper.updateByPrimaryKeySelective(product);
        }

        //创建订单
        Order order = new Order();
        order.setOrderAmount(product.getPrice().multiply(new BigDecimal(purchaseProductNum)));
        order.setOrderStatus(1);//待处理
        order.setReceiverName("xxx");
        order.setReceiverMobile("13311112222");
        order.setCreateTime(new Date());
        order.setCreateUser("xxx");
        order.setUpdateTime(new Date());
        order.setUpdateUser("xxx");
        orderMapper.insertSelective(order);

        //创建订单中的
        OrderItem orderItem = new OrderItem();
        orderItem.setOrderId(order.getId());
        orderItem.setProductId(product.getId());
        orderItem.setPurchasePrice(product.getPrice());
        orderItem.setPurchaseNum(purchaseProductNum);
        orderItem.setCreateUser("xxx");
        orderItem.setCreateTime(new Date());
        orderItem.setUpdateTime(new Date());
        orderItem.setUpdateUser("xxx");
        orderItemMapper.insertSelective(orderItem);
        return order.getId();
    }

ReentrantLock

   /**
     * 通过ReentrantLock锁解决并发问题
     */
    public Integer createOrderByReentrantLock() throws Exception {
        Product product = null;

        TransactionStatus transaction = platformTransactionManager.getTransaction(transactionDefinition);

        reentrantLock.lock();
        try {
            product = productMapper.selectByPrimaryKey(purchaseProductId);
            if (product == null) {
                throw new Exception("购买商品：" + purchaseProductId + "不存在");
            }

            //商品当前库存
            Integer currentCount = product.getCount();

            //校验库存
            if (purchaseProductNum > currentCount) {
                throw new Exception("商品" + purchaseProductId + "仅剩" + currentCount + "件，无法购买");
            }

            //计算剩余库存
            int leftCount = currentCount - purchaseProductNum;
            //更新库存
            product.setCount(leftCount);

            // 将提交的数据添加到数据库中.
            product.setUpdateTime(new Date());
            product.setUpdateUser("xxx");
            productMapper.updateByPrimaryKeySelective(product);

            platformTransactionManager.commit(transaction);
        } catch (Exception e) {
            platformTransactionManager.rollback(transaction);
            e.printStackTrace();
        } finally {
            reentrantLock.unlock();
        }

        TransactionStatus transaction1 = platformTransactionManager.getTransaction(transactionDefinition);

        //创建订单
        Order order = new Order();
        assert product != null;
        order.setOrderAmount(product.getPrice().multiply(new BigDecimal(purchaseProductNum)));
        order.setOrderStatus(1);
        order.setReceiverName("xxx");
        order.setReceiverMobile("13311112222");
        order.setCreateTime(new Date());
        order.setCreateUser("xxx");
        order.setUpdateTime(new Date());
        order.setUpdateUser("xxx");
        orderMapper.insertSelective(order);

        //创建订单中的
        OrderItem orderItem = new OrderItem();
        orderItem.setOrderId(order.getId());
        orderItem.setProductId(product.getId());
        orderItem.setPurchasePrice(product.getPrice());
        orderItem.setPurchaseNum(purchaseProductNum);
        orderItem.setCreateUser("xxx");
        orderItem.setCreateTime(new Date());
        orderItem.setUpdateTime(new Date());
        orderItem.setUpdateUser("xxx");
        orderItemMapper.insertSelective(orderItem);

        platformTransactionManager.commit(transaction1);

        return order.getId();
    }

分布式锁

在分布式集群的环境中，上面的单体锁会失效，此时需要通过分布式锁去处理并发问题。

数据库update锁

通过下面的sql给数据加锁，其它请求去更改数据时会被阻塞。

select... for update

Controller中的代码

    @Autowired
    private DistributeLockMapper distributeLockMapper;

    @RequestMapping("/test1")
    public String test1() throws Exception {
        log.info("进入方法..");
        DistributeLock lock = distributeLockMapper.selectLockByCode("demo");
        if (lock == null) {
            throw new Exception("找不到锁");
        }
        log.info("获取到了锁");

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "执行完成";
    }

mapper文件中的代码

    <select id="selectLockByCode" resultType="com.lock.distribute_lock.model.DistributeLock">
        select * from distribute_lock where business_code = #{code} for update;

    </select>

redis实现分布式锁

Redis实现分布式锁的核心命令：

SETNX key value

SETNX 命令的作用是：如果指定的 key 不存在，则创建并为其设置值，然后返回状态码 1；如果指定的 key 存在，则直接返回 0。如果返回值为 1，代表获得该锁；此时其他进程再次尝试创建时，由于 key 已经存在，则都会返回 0 ，代表锁已经被占用。

当获得锁的进程处理完成业务后，再通过 del 命令将该 key 删除，其他进程就可以再次竞争性地进行创建，获得该锁。

封装的RedisLock工具类

@Slf4j
public class RedisLock implements AutoCloseable{

    private RedisTemplate redisTemplate;

    private String key;

    private String value;

    private Long expireTime;

    public RedisLock(RedisTemplate redisTemplate, String key, Long expireTime) {
        this.redisTemplate = redisTemplate;
        this.key = key;
        this.value = UUID.randomUUID().toString();
        this.expireTime = expireTime;
    }

    /**
     * 获取分布式锁
     * @return
     */
    public boolean getLock() {
        RedisCallback<Boolean> redisCallback = redisConnection -> {
            //设置NX
            RedisStringCommands.SetOption setOption = RedisStringCommands.SetOption.ifAbsent();
            //设置过期时间
            Expiration expiration = Expiration.seconds(expireTime);
            //序列化key
            byte[] redisKey = redisTemplate.getKeySerializer().serialize(key);
            //序列化value
            byte[] redisValu = redisTemplate.getValueSerializer().serialize(value);
            //执行setNx操作
            Boolean result = redisConnection.set(redisKey, redisValu, expiration, setOption);
            return result;
        };

        //获取分布式锁
        Boolean lock  = (Boolean) redisTemplate.execute(redisCallback);

        return lock;
    }


    public boolean unLock() {
        String script = "if redis.call(\"get\",KEYS[1]) == ARGV[1] then\n" +
                "    return redis.call(\"del\",KEYS[1])\n" +
                "else\n" +
                "    return 0\n" +
                "end";
        RedisScript<Boolean> redisScript = RedisScript.of(script, Boolean.class);
        List<String> keys = Arrays.asList(key);
        Boolean result = (Boolean) redisTemplate.execute(redisScript, keys, value);
        log.info("释放锁的结果: " + result);
        return result;
    }


    @Override
    public void close() throws Exception {
        unLock();
    }
}

测试代码

    @Autowired
    private RedisTemplate redisTemplate;

    @RequestMapping("/redis-lock")
    public String redisLock() {
        log.info("redislock begin....");

        try (RedisLock redisLock = new RedisLock(redisTemplate, "redisKey", 30L)) {
            if (redisLock.getLock()) {
                log.info("获取了锁!!");
                //模拟业务操作
                Thread.sleep(15000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("redis lock end");

        return "方法执行完成";
    }

参考

  1. 深入解析 Redis 分布式锁原理

http://www.kler.cn/a/155747.html

相关文章：

  • 鸿蒙学习基本概念
  • docker配置代理解决不能拉镜像问题
  • 基于碎纸片的拼接复原算法及MATLAB实现
  • 【设计模式】行为型模式（二）：策略模式、命令模式
  • WebGIS三维地图框架--Cesium
  • ArkTs简单入门案例：简单的图片切换应用界面
  • MYSQL报错 [ERROR] InnoDB: Unable to create temporary file； errno: 0
  • WPF实战项目十九（客户端）：修改RestSharp的引用
  • 【Docker】容器数据持久化及容器互联
  • ThinkPHP的方法接收json数据问题
  • 【数据挖掘】国科大刘莹老师数据挖掘课程作业 —— 第三次作业
  • Vue3中teleport如何使用
  • 详解Spring对Mybatis等持久化框架的整合
  • LeetCode - 100. 相同的树 （C语言，二叉树，配图，简单）
  • 代理模式介绍（静态代理、jdk动态代理、cglib代理）
  • 栈和队列的OJ题——14.用栈实现队列
  • Azure Machine Learning - Azure AI 搜索中的索引器
  • 【限时免费】20天拿下华为OD笔试之【哈希集合】2023B-明明的随机数【欧弟算法】全网注释最详细分类最全的华为OD真题题解
  • Qt/QML编程学习之心得：如何添加资源文件到QML工程（十一）
  • 【书信，推荐信】【推荐节日活动】
  • QML中常见布局方法
  • 最小生成树（简单讲解，通俗易懂）
  • 笔迹检验（四）：笔迹检验的程序和方法
  • PyQt6 QComboBox下拉组合框控件
  • STM32串口接收不定长数据（接收中断+超时判断）
  • C++ Easyx 三子棋