几种分布式锁的简单实现

2021-07-08
0

分布式锁,顾名思义,在分布式环境中解决并发问题而采用的锁。这里要说的几种分布式锁的实现方式分别是基于数据库实现、基于redis实现以及基于ZooKeeper实现。

业务场景

商城抢单,下单后,判断商品库存是否大于0,若大于0,库存减1,不然,抢单失败。这里模拟的就是判断库存以及更新库存的逻辑。

表结构
字段 注释
id ID
goods_name 商品名称
goods_no 商品编号
goods_stock 商品剩余库存
version 记录版本号
create_time 创建时间
update_time 更新时间
实体类
/**
 * 商品库存信息实体类
 */
public class StockInfo {
    
    

    private String id;
    private String goodsName;
    private Integer goodsNo;
    private Integer goodsStock;
    private Integer version;
    private Date createTime;
    private Date updateTime;
    //get set方法不贴了
}

基于数据库实现

数据库乐观锁

之所以叫乐观锁,是因为它更倾向于程序执行的时候不会出现线程冲突的情况,比较乐观。而它的实现方式,其实类似于CAS(compare-and-swap)机制,大概流程如下图:
在这里插入图片描述
每次更新记录之前要比较拿到的version和库里的version是否一致,若一致表明没有其他线程操作该条记录,更新成功,version++;若不一致,表明当前线程在执行第2步时,有别的线程对该记录进行了操作,更新失败,执行失败策略。上代码:
service层:

/**
 * 数据库乐观锁和悲观锁实现分布式锁service
 */
@Service("distributeLockService_1")
public class DistributeLockService_1 {
    
    

    private static final Logger logger = LoggerFactory.getLogger(DistributeLockService_1.class);

    @Resource
    private StockInfoDao stockInfoDao;

    /**
     * 下单更新商品库存-乐观锁
     * @param goodsNo 商品编号
     * @return
     */
    public boolean order_1(Integer goodsNo){
    
    
        //如果是因为并发问题(有库存,版本号不对应)下单失败后,尝试下单操作的最大次数
        int count = 10;
        //是否继续尝试下单标志
        boolean flag = true;
        //当出现并发问题时,要继续尝试下单操作而不是直接放弃,所以这里用while循环,十次更新仍然失败则本次抢单失败。
        //这里对失败后的处理,是根据业务场景来决定的。
        while(count>0 && flag){
    
    
            count --;
            //根据商品编号查询该商品信息
            StockInfo stockInfo = stockInfoDao.selectByGoodsNo(goodsNo);
            //判断是否有库存.商品有库存时:
            if (stockInfo != null && stockInfo.getGoodsStock() > 0) {
    
    
                //更新库存。对比版本号
                stockInfo.setUpdateTime(new Date());
                int res = stockInfoDao.updateStockByGoodsNo(stockInfo);
                //更新成功,flag设问false,不再尝试减少库存操作。
                if(res > 0){
    
    
                    flag = false;
                    logger.info("抢单成功");
                    return true;
                }else{
    
    
                    //更新失败,则表明版本号对不上,即线程冲突,锁已被占用,继续尝试操作直到十次
                    logger.info("线程冲突,锁已被占用,本次获取锁失败,继续尝试抢单");
                }
            } else {
    
    
                //当商品库存不足时,不在继续尝试减少库存,下单失败
                flag = false;
                logger.info("商品{}库存不足", stockInfo.getGoodsName());
            }
        }
        return false;
    }

Dao层:

   /**
     * 根据商品编号查询商品信息
     * @param goodsNo
     * @return
     */
    public StockInfo selectByGoodsNo(Integer goodsNo);


    /**
     * 商品库存-1
     * @param stockInfo
     * @return
     */
    public int updateStockByGoodsNo(StockInfo stockInfo);

mapper映射文件:

 <!--  乐观锁 start  -->
    <select id="selectByGoodsNo" parameterType="java.lang.Integer" resultType="com.jason.distributelock.entity.StockInfo">
        select
        <include refid="Stock_Info_Column_List"></include>
        from stock_info
        where goods_no=#{goodsNo}
    </select>

    <update id="updateStockByGoodsNo" parameterType="com.jason.distributelock.entity.StockInfo">
        update stock_info
        set goods_stock = goods_stock-1,version = version+1
        where goods_no=#{goodsNo}
        and version = #{version}
    </update>
数据库悲观锁

数据库悲观锁,认为总是会出现线程不安全的问题,所以它会保证线程间的绝对同步。它是根据数据库行级锁原理实现的,即当某个事务正在操作某条记录时,不允许其他事务对该条记录进行update操作。在代码里当然是要开启事务的:
悲观锁Service:

/**
     * 下单更新商品库存-悲观锁
     * @param goodsNo 商品编号
     * @return
     * 使用悲观锁必须开启事务
     */
    @Transactional(rollbackFor = Exception.class)
    public boolean order_2(Integer goodsNo) {
    
    

        //根据商品编号查询库存信息,并锁住该行记录不让其他线程操作直到本次事务结束
        StockInfo stockInfo = stockInfoDao.selectByGoodsNoForUpdate(goodsNo);
        //判断该商品是否有库存
        if (stockInfo != null && stockInfo.getGoodsStock() > 0) {
    
    
            //更新库存
            stockInfo.setUpdateTime(new Date());
            stockInfoDao.updateStockByGoodsNoForUpdate(stockInfo);
            logger.info("抢单成功");
            return true;
        }else{
    
    
            logger.info("该商品{}没有库存了",stockInfo.getGoodsName());
        }
        return false;
    }

悲观锁Dao层:

/**
     * 根据商品编号查询商品信息
     * @param goodsNo
     * @return
     */
    public StockInfo selectByGoodsNoForUpdate(Integer goodsNo);


    /**
     * 商品库存-1
     * @param stockInfo
     * @return
     */
    public int updateStockByGoodsNoForUpdate(StockInfo stockInfo);

mapper文件:

<select id="selectByGoodsNoForUpdate" parameterType="java.lang.Integer" resultType="com.jason.distributelock.entity.StockInfo">
        select
        <include refid="Stock_Info_Column_List"></include>
        from stock_info
        where goods_no=#{goodsNo}
        for update
    </select>

    <update id="updateStockByGoodsNoForUpdate" parameterType="com.jason.distributelock.entity.StockInfo">
        update stock_info
        set goods_stock = goods_stock-1
        where goods_no=#{goodsNo}
    </update>

悲观锁在每次操作都会锁住该条记录,导致其他线程阻塞,这在高并发的情况下,系统压力会比较大。

Redis实现分布式锁

Redis实现分布式锁,是利用它的SETNX命令来实现的。这里将商品编号作为key尝试SETNX,详见代码及注释:
Service层:

   public boolean order(Integer goodsNo) throws Exception {
    
    
        String key = goodsNo + "";
        String value = UUID.randomUUID().toString();
        //获取锁失败后重试次数
        int count = 10;
        //是否继续获取锁标志
        boolean flag = true;
        while(count > 0 && flag){
    
    
            count --;
            //获取锁。这里用的是redis的setnx机制,当前key不存在才能设置成功,返回true,否则返回false
            if(redisTemplate.opsForValue().setIfAbsent(key,value)){
    
    
                try {
    
    
                    //获取锁成功,继续获取标志改为false
                    flag = false;
                    //查询商品库存信息
                    StockInfo stockInfo = stockInfoDao.selectByGoodsNoForNormal(goodsNo);
                    //判断是否有库存
                    if (stockInfo != null && stockInfo.getGoodsStock() > 0) {
    
    
                        //如果有库存,抢单成功,更新数据库
                        stockInfo.setUpdateTime(new Date());
                        stockInfoDao.updateStockByGoodsNoForNormal(stockInfo);
                        logger.info("抢单成功");
                        return true;
                    } else {
    
    
                        logger.info("库存不足");
                        return false;
                    }
                }catch (Exception e){
    
    
                    throw new Exception(e);
                }finally {
    
    
                    //无论什么情况,都要释放锁,否则会造成死锁。判断valu是否一直来确定释放的是否是当前持有的锁
                    if(value.equals(redisTemplate.opsForValue().get(key))) {
    
    
                        redisTemplate.delete(key);
                        logger.info("释放了锁");
                    }
                }
            }else{
    
    
                logger.info("线程冲突,获取锁失败");
            }
        }


        return false;
    }

DAO层:

/**
     * 根据商品编号查询商品信息
     * @param goodsNo
     * @return
     */
    public StockInfo selectByGoodsNoForNormal(Integer goodsNo);


    /**
     * 商品库存-1
     * @param stockInfo
     * @return
     */
    public int updateStockByGoodsNoForNormal(StockInfo stockInfo);

mapper文件:

<select id="selectByGoodsNoForNormal" parameterType="java.lang.Integer" resultType="com.jason.distributelock.entity.StockInfo">
        select
        <include refid="Stock_Info_Column_List"></include>
        from stock_info
        where goods_no=#{goodsNo}
    </select>

    <update id="updateStockByGoodsNoForNormal" parameterType="com.jason.distributelock.entity.StockInfo">
        update stock_info
        set goods_stock = goods_stock-1
        where goods_no=#{goodsNo}
    </update>

包依赖:

<!-- redis 依赖  -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <!-- redis 连接池依赖 -->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
            <version>${commons-pool2.version}</version>
        </dependency>

redis配置

#Redis以及连接池配置配置
spring.redis.host=127.0.0.1
spring.redis.port=6379
spring.redis.password=
spring.redis.timeout=2000

这样实现分布式锁,其实并不完善,当key设置完成拿到锁以后,如果此时系统宕机,那么锁是无法释放的,会造成死锁。所以,应当再设置一个key的有效期,过期后自动释放锁,而且加锁和设置有效期应当是一个原子操作,这里可以利用redis的set的一个多参数方法来实现:

set(key,value,nx,px,10000)

第一个为key,使用key来当锁名
第二个为value,传的是uid,唯一随机数,也可以使用本机mac地址 + uuid
第三个为NX,意思是SET IF NOT EXIST,即当key不存在时,进行set操作;若key已经存在,则不做任何操作
第四个为PX,意思是要给这个key加一个过期的设置,具体时间由第五个参数决定 第五个为time,代表key的过期时间,对应第四个参数 PX毫秒,EX秒
这样,既满足了互斥性,又不会造成死锁。

同样的,释放锁的时候,也是分两步,第一步判断value,第二部delete操作。那么同样要求这两个操作的原子性,可以用lua脚本来完美实现。对于lua脚本,小弟知之甚少,还有待学习。

ZooKeeper实现分布式锁

Zookeeper作分布式锁,是利用它节点互斥的原理,其实原理还是挺复杂的,做一个demo还好,生产环境容易产生一些bug。不过Zookeeper有一些成熟的包可以直接实现分布式锁,比方说牛逼的Curator,它封装了ZooKeeper的常用API,这里我用的就是它,贴代码:
pom引包:

<!--    zookeeper  start  -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>${zookeeper.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>${curator-framework.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>${curator-recipes}</version>
        </dependency>
        <!--    zookeeper end    -->

Dao层和Mapper文件代码和Redis实现分布式锁一样,就不贴了。
看Service层:

/**
 * Zookeeper实现分布式锁
 */
@Service
public class DistributeLockService_3 {
    
    

    private static final Logger logger = LoggerFactory.getLogger(DistributeLockService_3.class);

    @Resource
    private StockInfoDao stockInfoDao;

    /**
     * zookeeper链接客户端
     */
    @Autowired
    private CuratorFramework client;

    @Transactional(rollbackFor = Exception.class)
    public boolean order(Integer goodsNo) throws Exception {
    
    

        //看源码可知:path必须以“/”开头,不能以“/”结束
        String path = "/"+goodsNo+"-lock";
        //创建互斥锁
        InterProcessMutex lock = new InterProcessMutex(client, path);
        //3秒内尝试获取锁直到成功
        //这里要知道,zookeeper实现的分布式锁是有序的,先到先得。
        if ( lock.acquire(3L, TimeUnit.SECONDS) ){
    
    
            try{
    
    
                //查询商品库存信息
                StockInfo stockInfo = stockInfoDao.selectByGoodsNoForNormal(goodsNo);
                //判断是否有库存
                if(stockInfo != null && stockInfo.getGoodsStock() >0){
    
    
                    //如果有库存,减少库存,下单成功
                    stockInfo.setUpdateTime(new Date());
                    stockInfoDao.updateStockByGoodsNoForNormal(stockInfo);
                    logger.info("下单成功");
                    return true;
                }else{
    
    
                    //库存不足
                    logger.info("库存不足");
                }
            }finally{
    
    
                //释放锁
                lock.release();
            }
        }else{
    
    
            logger.info("未进入到锁");
        }

        return false;
    }
}

创建Zookeeper客户端Curator:

@SpringBootApplication
public class DistributeLockApplication {
    
    

    @Autowired
    private Environment env;

    /**
     * 创建zookeeper连接客户端
     * @return
     */
    @Bean
    public CuratorFramework curatorFramework(){
    
    
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString(env.getProperty("zk.host")).
                namespace(env.getProperty("zk.namespace")).retryPolicy(new RetryNTimes(5,1000)).build();
        client.start();
        return client;
    }

    public static void main(String[] args) {
    
    
        SpringApplication.run(DistributeLockApplication.class, args);
    }

}

在配置文件application.properties中配置zookeeper地址:

#ZooKeeper配置
zk.host=127.0.0.1:2181
zk.namespace=distribute_lock

实现很简单,但是zookeeper来实现分布式锁并不是很推荐,它的原理比较复杂,而且还要维护zookeeper集群,频繁地watch对集群的压力还是挺大的。

基于Redisson实现

写在了另一篇文章

对比总结

数据库悲观锁:比较适合并发量不高,读少写多,要求安全性比较高的场景。
数据库乐观所:适合读多写少的场景,但既然是基于数据库的锁,并发量也不应当太大,不然数据库压力会很大。
redis锁:是比较推荐的,原理简单,实现简单,性能好,安全。
zookeeper锁:上面说了,不是很推荐。如果不是原有的项目已经使用Zookeeper,同时锁的量级比较小的话,还是不用为妙。
redisson:简单安全性能好又很强大,推荐


点击查看完整代码

记录成长,热爱生活!

https://blog.csdn.net/qq_36798969/article/details/94721532

相关信息

评论