分布式锁实现的几种方式(DB,Redis,Zookeeper)

谁践踏了优雅 2022-12-31 03:24 132阅读 0赞

### 首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件： ###

**1.互斥性。在任意时刻，只有一个客户端能持有锁。  
2.不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。  
3.具有容错性。只要大部分的Redis节点正常运行，客户端就可以加锁和解锁。  
4.解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了**。

### 应用场景：订单模块新增一条订单，商品模块减少库存，共用一个数据库 ###

**不加锁代码：**

/**
         * 减少库存-不加锁-会存在超卖问题
         *
         * @param productId
         * @return
         */
        @Transactional
        @Override
        public Pair<Boolean, String> reductStock(Integer productId) {
            Integer stock = this.productMapper.getStock(productId);
            if (stock <= 0) {
                return Pair.of(false, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】库存不足,剩余库存{0}", stock));
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            this.productMapper.reductStock(productId);
            return Pair.of(true, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】减少库存成功,剩余库存{0}", this.productMapper.selectByPrimaryKey(productId).getStock()));
        }

<select id="getStock" resultType="java.lang.Integer">
             select stock from product where id=#{productId}
    </select>

<update id="reductStock">
            update product set stock=stock-1 where id=#{productId}
    </update>

### **方式一（数据库-for update悲观锁）：** ###

**        /\*\*  
         \* 总结几个点：  
         \* 1.for update 锁必须使用事务，不然起不到真正锁的效果，数据库默认的写语句都是执行即提交  
         \* 2.for update 锁是行锁，查询语句必须走主键索引，否则可能锁掉其他数据或者整个表  
         \* 3.for update加锁后，其他线程或进程中数据库客户端执行语句和java客户端执行语句都会进入等待状态  
         \* 4.for update加锁后，还没来得及解锁程序异常时，事务会回滚，释放锁  
         \*/**

/**
         * 减少库存-数据库for update锁-悲观锁
         *
         * @param productId
         * @return
         */
        @Transactional
        @Override
        public Pair<Boolean, String> reductStockByLock(Integer productId) {
            //加锁
            Integer stock = this.productMapper.getStockByLock(productId);
            logger.info("【线程" + Thread.currentThread().getId() + "】加锁成功。。。");
            if (stock <= 0) {
                return Pair.of(false, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】库存不足,剩余库存{0}", stock));
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //如果加锁后程序遇到异常，事务会释放当前锁资源
            //Integer num = 1 / 0;
            //解锁
            this.productMapper.reductStock(productId);
            logger.info("【线程" + Thread.currentThread().getId() + "】解锁成功。。。");
            return Pair.of(true, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】减少库存成功,剩余库存{0}", this.productMapper.selectByPrimaryKey(productId).getStock()));
        }

<select id="getStockByLock" resultType="java.lang.Integer">
            select stock from product where id=#{productId} for update
    </select>

### **方式一（数据库-乐观锁）：** ###

/**
         * 减少库存-数据库version锁-乐观锁
         *
         * @param productId
         * @return
         */
        @Transactional
        @Override
        public Pair<Boolean, String> reductStockByVersionLock(Integer productId) {
            Product product = this.productMapper.selectByPrimaryKey(productId);
            Integer stock = product.getStock();
            String version = product.getLockVersion();
            if (product.getStock() <= 0) {
                return Pair.of(false, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】,版本:" + version + ",库存不足,剩余库存{0}", stock));
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //更新的时候条件where要带上version
            Integer result = this.productMapper.reductStockByVersionLock(productId, version);
            if (null == result || result <= 0) {
                return Pair.of(true, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】,版本:" + version + ",减少库存失败,剩余库存{0}", this.productMapper.selectByPrimaryKey(productId).getStock()));
            }
            return Pair.of(true, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】,版本:" + version + ",减少库存成功,剩余库存{0}", this.productMapper.selectByPrimaryKey(productId).getStock()));
        }

<update id="reductStockByVersionLock">
           update product set stock=stock-1,lock_version=lock_version+1 where id=#{productId} and lock_version=#{version}
    </update>

### **方式二（Redis-Lua实现）：** ###

**错误版本：**

/**
         * 加锁
         *
         * @param lockKeyEnum 锁类型
         * @param lockDataId  锁对应的业务表数据id
         */
        public Boolean lock(LockKeyEnum lockKeyEnum, String lockDataId) {
            Jedis jedis = null;
            String lockKey = lockKeyEnum.getCode() + "_" + lockDataId;
            try {
                jedis = jedisPool.getResource();
                if (null == jedis) {
                    return false;
                }
                //1表示成功 0表示失败（该key已存在）
                if (jedis.setnx(lockKey, lockDataId).equals(0)) {
                    return false;
                }
                jedis.expire(lockKey, 10);
                return true;
            } catch (Exception e) {
                logger.error("加锁异常。。。", e);
                //手动解锁
                this.unLock(jedis, lockKeyEnum, lockDataId);
            } finally {
                //释放资源
                jedis.close();
            }
            return false;
        }
    
    
        /**
         * 解锁
         *
         * @param lockKeyEnum 锁类型
         * @param lockDataId  锁对应的业务表数据id
         */
        public Boolean unLock(Jedis jedis, LockKeyEnum lockKeyEnum, String lockDataId) {
            jedis.del(lockKeyEnum.getCode() + "_" + lockDataId);
            return true;
        }

**错误点1（造成死锁）：**

加锁时：

如果执行到jedis.setnx整个程序挂掉了，这里指的是程序直接退出，try catch没有用，会导致死锁

![20210103222711264.png][]

**改进1：**

这里让判断赋值和设置过期时间一起执行，原子性操作

![20210103224016992.png][]

**错误点2（删除掉别人的锁）：**

解锁时：

假设这里有ABC三个客户端请求排队，这时A客户端抢到了锁设置过期时间为10秒钟，如果这里执行的时间比较长，当第11秒钟的时候，（客户端A的）锁过期，

客户端B马上抢到锁并上锁，这时客户端A恰好执行完成，进行解锁操作，就会把（客户端B的）锁删除，这时客户端B处于裸奔状态，客户端C马上抢到锁并上锁，

这时问题来了，客户端B和客户端C的操作不是原子性，会出现互相抢资源，出现超卖的情况出现

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWNoYW5nMTk5NTA3MDM_size_1_color_FFFFFF_t_70][]

**改进2**

给每一次客户端请求都加上version版本标记，解锁的时候判断是不是自己客户端请求的锁

加锁时：

![20210103223829593.png][]

解锁时：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWNoYW5nMTk5NTA3MDM_size_1_color_FFFFFF_t_70 1][]

**错误点3（非原子操作删除掉别人的锁）：**

改进了2次后，似乎没问题了，但其实还是有一个隐藏的问题，看这段解锁代码，

假设客户端A加锁后，正常执行完成，在解锁的时候，判断版本这一句代码执行完的时候，（客户端A的）锁正好过期，然后（客户端B的）锁加上，

客户端A的解锁操作继续执行，这样客户端B的锁就被删除掉，客户端B就变成裸奔状态了

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWNoYW5nMTk5NTA3MDM_size_1_color_FFFFFF_t_70 1][]

**改进3**

为了解决解锁时不是原子性操作的问题，解锁时必须引入lua语法来保证该操作是原子性的操作

所以最终的代码是：

package com.study.utils;
    
    import com.study.enums.LockKeyEnum;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisPool;
    
    import java.util.Collections;
    import java.util.UUID;
    
    
    @Component
    public class RedisLockByLua {
    
        private static final Logger logger = LoggerFactory.getLogger(RedisLockByLua.class);
    
    
        /**
         * 加锁-返回状态
         */
        private static final String LOCK_SUCCESS = "OK";
    
        /**
         * 解锁-返回状态
         */
        private static final Long RELEASE_SUCCESS = 1L;
    
        /**
         * 设置-如果不存在就添加
         */
        private static final String SET_IF_NOT_EXIST = "NX";
        /**
         * 设置-过期时间
         */
        private static final String SET_WITH_EXPIRE_TIME = "PX";
    
        /**
         * 过期时间(10秒钟)
         */
        private static final Long EXPIRE_TIME = 20 * 1000L;
    
        /**
         * 加锁
         *
         * @param lockKeyEnum
         *         锁类型
         * @param lockDataId
         *         锁对应的业务表数据id
         * @param lockVersion
         *         锁对应的版本（UUID）
         */
        public Boolean lock(Jedis jedis, LockKeyEnum lockKeyEnum, String lockDataId, String lockVersion) {
            String lockKey = lockKeyEnum.getCode() + "_" + lockDataId;
            try {
                while (true) {
                    String result = jedis.set(lockKey, lockVersion, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, EXPIRE_TIME);
                    if (LOCK_SUCCESS.equals(result)) {
                        //加锁成功，跳出循环
                        return true;
                    }
                    logger.info("【线程{}】【版本:{}】等待加锁。。。", Thread.currentThread().getId(), lockVersion);
                }
            } catch (Exception e) {
                logger.error("【线程{" + Thread.currentThread().getId() + "}】【版本:{" + lockVersion + "}】加锁异常。。。", e);
                //手动解锁
                this.unLock(jedis, lockKeyEnum, lockDataId, lockVersion);
                return false;
            }
        }
    
        /**
         * 解锁
         *
         * @param lockKeyEnum
         *         锁类型
         * @param lockDataId
         *         锁对应的业务表数据id
         * @param lockVersion
         *         锁对应的版本
         */
        public Boolean unLock(Jedis jedis, LockKeyEnum lockKeyEnum, String lockDataId, String lockVersion) {
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            Object result = jedis.eval(script, Collections.singletonList(lockKeyEnum.getCode() + "_" + lockDataId), Collections.singletonList(lockVersion));
            if (RELEASE_SUCCESS.equals(result)) {
                return true;
            }
            return false;
        }
    
    }

### **方式三（Redission实现）：** ###

**这里有一篇文章详解了Redisson原理机制，大概Redisson内部加锁和解锁都用了Lua实现，保证原子性操作**

[https://blog.csdn.net/asd051377305/article/details/108384490][https_blog.csdn.net_asd051377305_article_details_108384490]

/**
         * 减少库存-Redis-Redisson锁
         *
         * @param productId
         * @return
         */
        @Override
        public Pair<Boolean, String> reduceStockByRedisson(Integer productId) throws Exception {
            RLock rLock = null;
            try {
                //1.配置
                String address = "redis://" + REDIS_HOSTNAME + ":" + REDIS_PORT;
                Config config = new Config();
                config.useSingleServer().setAddress(address).setDatabase(0);
                //2.构建RedissonClient
                RedissonClient redissonClient = Redisson.create(config);
                String lockKey = LockKeyEnum.Product_Stock.getCode() + "_" + productId;
                rLock = redissonClient.getLock(lockKey);
                //等待获取锁的最大等待时间，超过这个值，则认为获取锁失败
                Long waitTimeOut = 30L;
                //锁的超时时间，超过这个时间锁会自动失效(设置的时间比业务处理时间大，保证有效期内业务能处理完成)
                Long leaseTime = 10L;
                logger.info("【线程{}】准备获取锁。。。", Thread.currentThread().getId());
                /**
                 * 1.获取不到锁时会阻塞,内部实现了while(true)循环
                 * 2.1如果我们指定了锁的超时时间 leaseTime，就发送给redis执行脚本，进行占锁，默认超时就是我们制定的时间，不会自动续期；
                 * 2.2如果我们未指定锁的超时时间 leaseTime，就使用 lockWatchdogTimeout = 30 * 1000 【看门狗默认时间】
                 * 只要占锁成功，就会启动一个定时任务【重新给锁设置过期时间，新的过期时间就是看门狗的默认时间】,每隔10秒都会自动再续成30秒；
                 */
                //3.加锁
                //Boolean result = rLock.tryLock(waitTimeOut, TimeUnit.SECONDS);
                Boolean result = rLock.tryLock(waitTimeOut, leaseTime, TimeUnit.SECONDS);
                if (result) {
                    logger.info("【线程{}】获取锁成功，开始业务。。。", Thread.currentThread().getId());
                    Product product = this.productMapper.selectByPrimaryKey(productId);
                    Integer stock = product.getStock();
                    if (stock <= 0) {
                        return Pair.of(false, MessageFormat.format("【线程{0}】库存不足,剩余库存{1}", Thread.currentThread().getId(), stock));
                    }
                    Thread.sleep(5000);
                    this.productMapper.reductStock(productId);
                    logger.info("【线程{}】业务执行完成。。。", Thread.currentThread().getId());
                    return Pair.of(true, MessageFormat.format("【线程{0}】减少库存成功,剩余库存{1}", Thread.currentThread().getId(), this.productMapper.selectByPrimaryKey(productId).getStock()));
                } else {
                    logger.info("【线程{}】获取锁失败。。。", Thread.currentThread().getId());
                    return Pair.of(false, MessageFormat.format("【线程{0}】获取锁失败。。。", Thread.currentThread().getId()));
                }
            } catch (Exception e) {
                logger.error("【线程" + Thread.currentThread().getId() + "】减少库存异常。。。", e);
                throw e;
            } finally {
                //4.最后都要解锁(判断1.解锁状态2.解锁当前线程的锁)
                if (null != rLock && rLock.isLocked() && rLock.isHeldByCurrentThread()) {
                    logger.info("【线程{}】解锁。。。", Thread.currentThread().getId());
                    rLock.unlock();
                }
            }
        }

### **方式四（Zookeeper实现）：** ###

package com.study.utils;
    
    
    import com.study.enums.LockKeyEnum;
    import org.I0Itec.zkclient.IZkDataListener;
    import org.I0Itec.zkclient.ZkClient;
    import org.apache.commons.lang3.StringUtils;
    import org.apache.zookeeper.*;
    import org.apache.zookeeper.data.Stat;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    
    import java.io.IOException;
    import java.util.Collections;
    import java.util.Comparator;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.stream.Collectors;
    
    
    public class ZookeeperLock implements Lock {
    
        public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
            //ZookeeperLock lock = new ZookeeperLock(LockKeyEnum.Product_Stock.getCode() + "_1");
            //lock.lock();
            ...执行业务
            //Thread.sleep(1000);
            //lock.unlock();
        }
    
    
        private static final Logger logger = LoggerFactory.getLogger(ZookeeperLock.class);
    
        @Value("${zookeeper.url}")
        private String zkUrl;
        /**
         * 锁-key
         */
        private String lockKey;
    
        /**
         * zk客户端
         */
        private ZkClient zk;
    
    
        /**
         * 客户端连接服务器超时时间(毫秒)
         */
        private static final Integer sessionTimeOut = 30000;
    
        /**
         * 计数器
         */
        private CountDownLatch latch;
    
    
        /**
         * 锁-根目录地址
         */
        private static final String rootNodePath = "/myLock";
    
        /**
         * 当前锁节点
         */
        private String nowNodePath;
    
        /**
         * 当前锁节点 等待释放的前一个节点
         */
        private String waitNodePath;
    
    
        public ZookeeperLock(String _lockKey) {
            this.lockKey = _lockKey;
            try {
                zk = new ZkClient(StringUtils.isEmpty(zkUrl) ? "192.168.247.129:2181" : zkUrl);
                if (!zk.exists(rootNodePath)) {
                    //创建根节点(持久节点)
                    zk.createPersistent(rootNodePath);
                }
            } catch (Exception e) {
                logger.error("初始化zk客户端异常。。。", e);
                if (null != zk) {
                    zk.close();
                }
            }
        }
    
    
        /**
         * 获取锁
         */
        @Override
        public void lock() {
            if (this.tryLock()) {
                logger.info("【线程{}】获取锁成功。。。", Thread.currentThread().getId());
                return;
            } else {
                //等待锁
                this.waitForLock();
                //递归重新尝试获取锁，一旦监听到上一个锁线程释放了锁，再次重新获取锁
                logger.info("【线程{}】重新尝试获取锁。。。", Thread.currentThread().getId());
                lock();
            }
        }
    
        /**
         * 等待获取锁
         *
         * @return
         */
        private Boolean waitForLock() {
            try {
                logger.info("【线程{}】等待获取锁。。。", Thread.currentThread().getId());
                //注册监听此节点的变化
                IZkDataListener iZkDataListener = new IZkDataListener() {
                    @Override
                    public void handleDataChange(String dataPath, Object data) throws Exception {
    
                    }
    
                    @Override
                    public void handleDataDeleted(String dataPath) throws Exception {
                        logger.info("【线程{}】【节点{}】监控到被释放删除。。。", Thread.currentThread().getId(), dataPath);
                        if (null != latch) {
                            latch.countDown();
                        }
                    }
                };
                //子节点加入Watcher监控
                zk.subscribeDataChanges(waitNodePath, iZkDataListener);
                if (zk.exists(waitNodePath)) {
                    logger.info("【线程" + Thread.currentThread().getId() + "】【节点{}】阻塞，【节点{}】等待解锁。。。", nowNodePath, waitNodePath);
                    this.latch = new CountDownLatch(1);
                    //这里阻塞，一直等待其他线程释放锁
                    this.latch.await();
                    logger.info("【线程" + Thread.currentThread().getId() + "】【节点{}】阻塞解除，【节点{}】解锁。。。", nowNodePath, waitNodePath);
                }
                //子节点释放Watcher监控
                zk.unsubscribeDataChanges(waitNodePath, iZkDataListener);
                return true;
            } catch (Exception e) {
                logger.error("等待获取锁异常。。。", e);
                throw new RuntimeException("等待获取锁异常。。。", e);
            }
        }
    
        /**
         * 它表示用来尝试获取锁，如果获取成功，则返回true；如果获取失败（即锁已被其他线程获取），则返回false，也就是说，这个方法无论如何都会立即返回（在拿不到锁时不会一直在那等待）
         *
         * @return
         */
        @Override
        public boolean tryLock() {
            try {
                String splitStr = "_";
                //这个判断很重要,如果第一次进来tryLock,当前线程节点未指定,创建当前节点,不判断的话就会死循环,当前线程节点永远为最新产生的+1的节点
                if (StringUtils.isEmpty(nowNodePath)) {
                    nowNodePath = zk.createEphemeralSequential(rootNodePath + "/" + lockKey + splitStr, "lock");
                }
                //获取所有子节点
                List<String> allChildrenList = zk.getChildren(rootNodePath);
                //筛选出当前业务锁对应的子节点
                List<String> nowChildrenList = allChildrenList.stream().filter(a -> a.contains(lockKey)).collect(Collectors.toList());
                //排序，升序
                Collections.sort(nowChildrenList);
                //节点内容:  product_stock_1_0000000001
                //如果当前锁节点是最小的节点
                if (nowNodePath.equals(rootNodePath + "/" + nowChildrenList.get(0))) {
                    logger.info("【线程:{}】【节点:{}】为最小节点,获取到锁", Thread.currentThread().getId(), nowNodePath);
                    return true;
                }
                logger.info("【线程:{}】【节点:{}】不是最小节点,等待获取锁", Thread.currentThread().getId(), nowNodePath);
                //如果当前锁节点不是最小的节点，找到比自己小1的节点
                List<String> smallerNodeList = nowChildrenList.stream().filter(a -> Long.valueOf(a.substring(a.lastIndexOf(splitStr) + 1)) < Long.valueOf(nowNodePath.substring(nowNodePath.lastIndexOf(splitStr) + 1))).collect(Collectors.toList());
                waitNodePath = rootNodePath + "/" + smallerNodeList.get(smallerNodeList.size() - 1);
            } catch (Exception e) {
                logger.error("尝试获取锁异常。。。", e);
                throw new RuntimeException("尝试获取锁异常。。。", e);
            }
            return false;
        }
    
        /**
         * 这个方法在拿不到锁时会等待一定的时间，在时间期限之内如果还拿不到锁，就返回false，同时可以响应中断。如果一开始拿到锁或者在等待期间内拿到了锁，则返回true
         *
         * @param time
         * @param unit
         * @return
         * @throws InterruptedException
         */
        @Override
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
            return false;
        }
    
        /**
         * 解锁
         */
        @Override
        public void unlock() {
            try {
                //删除当前节点
                if (zk.exists(nowNodePath)) {
                    zk.delete(nowNodePath);
                }
                nowNodePath = null;
                //释放zk客户端连接
                zk.close();
                logger.info("【线程{}】解锁成功。。。", Thread.currentThread().getId());
            } catch (Exception e) {
                logger.error("解锁异常。。。", e);
                throw new RuntimeException("解锁异常。。。", e);
            }
        }
    
        /**
         * 返回绑定到此 Lock 实例的新 Condition 实例
         *
         * @return
         */
        @Override
        public Condition newCondition() {
            return null;
        }
    
        /**
         * 如果当前线程未被中断，则获取锁，可以响应中断
         *
         * @throws InterruptedException
         */
        @Override
        public void lockInterruptibly() throws InterruptedException {
    
        }
    
    }

**简单总结：**

**从理解的难易程度角度（从低到高）数据库 > 缓存 > Zookeeper**

**从性能角度（从高到低）缓存 > Zookeeper >= 数据库**

**从可靠性角度（从高到低）Zookeeper > 缓存 > 数据库**

[20210103222711264.png]: /images/20221120/e72cbd7fdb9941b39aafb2154d658bfc.png
[20210103224016992.png]: /images/20221120/e10b82e443ad46138727d842653f151d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWNoYW5nMTk5NTA3MDM_size_1_color_FFFFFF_t_70]: /images/20221120/bd0dbaad305b4b919d27aee06c5799e1.png
[20210103223829593.png]: /images/20221120/9dee9e46c61a45b0bfc8ebb093dd7d19.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWNoYW5nMTk5NTA3MDM_size_1_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210103224445501.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWNoYW5nMTk5NTA3MDM=,size_1,color_FFFFFF,t_70
[https_blog.csdn.net_asd051377305_article_details_108384490]: https://blog.csdn.net/asd051377305/article/details/108384490