欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

【连载】redis库存操作,分布式锁的四种实现方式[三]--基于Redis watch机制实现分布式锁

程序员文章站 2022-03-24 12:22:11
一、redis的事务介绍 1、 Redis保证一个事务中的所有命令要么都执行,要么都不执行。如果在发送EXEC命令前客户端断线了,则Redis会清空事务队列,事务中的所有命令都不会执行。而一旦客户端发送了EXEC命令,所有的命令就都会被执行,即使此后客户端断线也没关系,因为Redis中已经记录了所有 ......

一、redis的事务介绍

1、 redis保证一个事务中的所有命令要么都执行,要么都不执行。如果在发送exec命令前客户端断线了,则redis会清空事务队列,事务中的所有命令都不会执行。而一旦客户端发送了exec命令,所有的命令就都会被执行,即使此后客户端断线也没关系,因为redis中已经记录了所有要执行的命令。

2、 除此之外,redis的事务还能保证一个事务内的命令依次执行而不被其他命令插入。试想客户端a需要执行几条命令,同时客户端b发送了一条命令,如果不使用事务,则客户端b的命令可能会插入到客户端a的几条命令中执行。如果不希望发生这种情况,也可以使用事务。

3、 若一个事务中有多条命令,若有一条命令错误,事务中的所有命令都不会执行。若在执行阶段有命令执行错误,其他的命令也会正确的执行,需要注意。

4、与mysql的事务不同,redis的事务执行中时不会回滚的,哪怕出现错误,之前已经执行的命令结果也不会回滚。

二、redis watch介绍

1、 watch命令可以监控一个或多个键,一旦其中有一个键被修改(或删除),之后的事务就不会执行。监控一直持续到exec命令(事务中的命令是在exec之后才执行的,所以在multi命令后可以修改watch监控的键值)

2、watch一般配合事务使用

例:启动一个线程,连接redis,监控key watchkeytest,sleep10s模拟业务逻辑处理,此时再启动另一个进程去修改该key的值,那么当前线程就会返回null

/**
 * @author lijunjun
 * @date 2018/12/10
 */
public class test {

    private static jedis jedis;

    static {
        jedis = new jedis("192.168.10.109", 6379);
        jedis.auth("aaa@leadeon.cn");
        jedis.sadd("watchkeytest", "290");
    }

    public static void main(string[] args) {

        jedis.watch("watchkeytest");

        system.out.println("开始监控key: watchkeytest");

        transaction transaction = jedis.multi();

        try {
            // sleep 10秒,模拟业务逻辑处理
            thread.sleep(1000);
        } catch (interruptedexception e) {
            e.printstacktrace();
        }

        system.out.println("开始获取key: watchkeytest");
        transaction.sismember("watchkeytest", "290");

        list<object> result = transaction.exec();
        system.out.println("执行结果:" + result);
        jedis.disconnect();
    }
}

启动另一个进程,修改同一个key

public class test2 {

    public static void main(string[] args) {

        jedis jedis = new jedis("192.168.10.109", 6379);
        jedis.auth("common@leadeon.cn");
        long result = jedis.sadd("watchkeytest", "358");
        system.out.println(result);
        jedis.disconnect();
    }
}

此时,进程1就会返回null
【连载】redis库存操作,分布式锁的四种实现方式[三]--基于Redis watch机制实现分布式锁

若在进程1执行期间,该key没有被其他进程修改,则返回正确的值。

三、实现思路

基于以上介绍的redis的事务以及watch机制,我们可以做分布式锁处理,即在分布式系统中,高并发情况下,一个线程watch相应的key后,其他进程若修改了key,则该进程所在的事务就不执行,返回null,我们可以增加重试机制,来做库存操作

四、业务代码实现

采用watch机制,做乐观锁处理,重试三次,三次返回均未成功,则接口返回失败
    /**
     * 减库存(基于redis watch机制实现)
     *
     * @param trace 请求流水
     * @param stockmanagereq(stockid、decrnum)
     * @return -1为失败,大于-1的正整数为减后的库存量,-2为库存不足无法减库存
     */
    @override
    @apioperation(value = "减库存", notes = "减库存")
    @requestmapping(value = "/decrbystock", method = requestmethod.post, consumes = mediatype.application_json_utf8_value, produces = mediatype.application_json_utf8_value)
    public int decrbystock(@requestheader(name = "trace") string trace, @requestbody stockmanagereq stockmanagereq) {

        long starttime = system.currenttimemillis();

        logger.reqprint(log.cache_sign, log.cache_request, trace, "decrbystock", json.tojsonstring(stockmanagereq));

        int res = 0;
        string stockid = stockmanagereq.getstockid();
        integer decrnum = stockmanagereq.getdecrnum();

        boolean decrbystock = false;

        try {
            if (null != stockid && null != decrnum) {

                stockid = prefix + stockid;

                // 采用watch机制,做乐观锁处理,重试三次,三次返回均未成功,则接口返回失败
                for (int i = 0; i < try_count; i++) {
                    integer decrbystockres = decrbystock(stockid, decrnum, trace);

                    // 更新库存时key对应的value发生变更,重试
                    if (decrbystockres != -1) {
                        res = decrbystockres;
                        decrbystock = true;
                        break;
                    }
                }

                if (!decrbystock) {
                    res = -2;
                    logger.info("本次请求减库存失败!decrbystockfailure=1");
                }
            }
        } catch (exception e) {
            logger.error(trace, "decr sku stock failure.", e);
            res = -1;
        } finally {
            logger.respprint(log.cache_sign, log.cache_response, trace, "decrbystock", system.currenttimemillis() - starttime, string.valueof(res));
        }
        return res;
    }

    /**
     * 减库存逻辑
     *
     * @param stockid 库存id
     * @param decrnum 减少的量
     * @return 减库存结果(-1:表示更新库存时key对应的value发生变更,即提示调用方重试;-2: 库存不够减,售罄;其它值表示减库存后的值)
     */
    private integer decrbystock(string stockid, int decrnum, string trace) {

        response<long> v = null;
        list<object> result = null;

        try (jedis jedis = jedispool.getwriteresource()) {

            if (!jedis.select(0).equals("ok")) {
                logger.error(trace, "减库存,本次请求未获取到jedis连接!");
                return -1;
            }

            jedis.watch(stockid);

            // redis 减库存逻辑
            string vstock = jedis.get(stockid);

            long realv = 0l;

            if (stringutils.isnotempty(vstock)) {
                realv = long.parselong(vstock);
            }
            //库存数  大于等于 要减的数目,则执行减库存
            if (realv < decrnum) {

                return -2;
            }

            transaction transaction = jedis.multi();

            v = transaction.decrby(stockid, decrnum);

            result = transaction.exec();
        }

        return (result == null || result.isempty()) ? -1 : v.get().intvalue();
    }

五、ab压测及分析
同样的,我们以5000的请求量100的并发量来压、tps在640左右,比zk做分布式锁来看,提升了20倍的性能,比redisson分布式锁提升了2倍,性能提升不大

【连载】redis库存操作,分布式锁的四种实现方式[三]--基于Redis watch机制实现分布式锁

同时我们发现,5000个请求,有4561个请求失败,我们看下日志统计,有多少请求没有成功执行事务

【连载】redis库存操作,分布式锁的四种实现方式[三]--基于Redis watch机制实现分布式锁

也是4561,说明有4561个事务没有成功执行,并不是运行错误。

六、总结

watch可以用来控制对redis的操作同步执行,但失败的几率较大,用该机制做抢购的业务还行,但对redis操作结果依赖较强的业务来说,不太适用,下一篇我们讲下终极解决方案,适用redis的lua脚本编程,变相的实现分布式锁。