Java基于redis實(shí)現分布式鎖

目錄

為了保證一個(gè)在高并發(fā)存場(chǎng)景下只能被同一個(gè)線(xiàn)程操作，java并發(fā)處理提供ReentrantLock或Synchronized進(jìn)行互斥控制。但是這僅僅對單機環(huán)境有效。我們實(shí)現分布式鎖大概通過(guò)三種方式。

• redis實(shí)現分布式鎖
• 數據庫實(shí)現分布式鎖
• zk實(shí)現分布式鎖

實(shí)際上這三種和java對比看屬于一類(lèi)。都是屬于程序外部鎖。

原理剖析

上述三種分布式鎖都是通過(guò)各自為依據對各個(gè)請求進(jìn)行上鎖，解鎖從而控制放行還是拒絕。redis鎖是基于其提供的setnx命令。
setnx當且僅當key不存在。若給定key已經(jīng)存在，則setnx不做任何動(dòng)作。setnx是一個(gè)原子性操作。
和數據庫分布式相比，因為redis內存輕量。所以redis分布式鎖性能更好

實(shí)現

原理很簡(jiǎn)單。結合springboot項目我們實(shí)現一套通過(guò)注解形式對接口進(jìn)行庫存上鎖案例進(jìn)行理解

編寫(xiě)注解

我們編寫(xiě)注解。方便我們在接口上添加注解提供攔截信息

/**
 * @author 張新華
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface StockLock {

    /**
     * @author zxhtom
     * @Description 鎖key的前綴
     * @Date 15:25 2020年03月25日, 0025
     * @Param []
     * @return java.lang.String
     */
    String prefix() default "";
    /**
     * @author zxhtom
     * @Description key的分隔符
     * @Date 15:27 2020年03月25日, 0025
     * @Param []
     * @return java.lang.String
     */
    String delimiter() default ":";
}

/**
 * @author 張新華
 */
@Target({ElementType.PARAMETER , ElementType.METHOD , ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface StockParam {
    /**
    * @author zxhtom
    * @Description 組成key
    * @Date 11:11 2020年03月26日, 0026
    * @Param []
    * @return java.lang.String[]
    */
    String[] names() default {""};
}

攔截器攔截

redis分布式鎖實(shí)現的關(guān)鍵就是攔截器的編寫(xiě)。上面的注解只是為了實(shí)現攔截的一個(gè)輔助。

@Around("execution(public * *(..)) && @annotation(com.ay.framework.order.redis.product.StockLock)")

• 通過(guò)springboot的Around進(jìn)行針對StockLock注解的攔截。通過(guò)攔截我們可以獲取到攔截的方法、參數、及需要的鎖的參數。
• 我們獲取到需要鎖的名稱(chēng)這里叫做【a】之后通過(guò)redis的原子性操作對該key進(jìn)行遞減操作。
• 為了方便我們在削減庫存的時(shí)候可以對庫存進(jìn)行更新操作。我們在遞減庫存前還需要借助于另一把鎖。 這一把鎖我們叫做【a_key】
• 換句話(huà)說(shuō)我們接口想訪(fǎng)問(wèn)就必須獲取【a】鎖，拿到【a】鎖需要減少庫存。減少庫存之前需要獲取【a_key】鎖。
• 拿到鎖之后處理完邏輯之后我們需要釋放對應鎖。

RedisAtomicLong entityIdCounter = new RedisAtomicLong(lockKey, redisTemplate.getConnectionFactory());
    if (redisTemplate.hasKey(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey)) {
        //表示lockKey的庫存信息有變動(dòng)。此時(shí)無(wú)法進(jìn)行交易
        throw new BusinessException("庫存變動(dòng)。暫無(wú)法交易");
    }
    Long increment = entityIdCounter.decrementAndGet();
    if (increment >= 0) {
        try {
            Object proceed = pjp.proceed();
        } catch (Throwable throwable) {
            //所占資源需要釋放回資源池
            while (!redisLock.tryGetLock(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey, "")) {

            }
            //表示lockKey的庫存信息有變動(dòng)。此時(shí)無(wú)法進(jìn)行交易
            long l = entityIdCounter.incrementAndGet();
            if (l < 1) {
                redisTemplate.opsForValue().set(lockKey,1);
            }
            redisLock.unLock(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey);
            throwable.printStackTrace();
        }
    } else {
        redisTemplate.opsForValue().set(lockKey,0);
        throw new BusinessException("庫存不足！無(wú)法操作");
    }

因為我們上鎖就需要釋放鎖。但是程序在中途處理業(yè)務(wù)是發(fā)生異常導致沒(méi)有走到釋放鎖的步驟。這個(gè)時(shí)候就導致我們的分布式鎖一直被鎖。俗稱(chēng)【死鎖】。為了避免這種場(chǎng)景的發(fā)生。我們常常在上鎖的時(shí)候給一個(gè)有效期。有效期已過(guò)自動(dòng)釋放鎖。這個(gè)特性恰好和redis的過(guò)期策略不摩爾和。

上述提及工具

RedisLock

public Boolean tryGetLock(String key , String value) {
    return tryGetLock(key, value, -1, TimeUnit.DAYS);
}
public Boolean tryGetLock(String key , String value, Integer expire) {
    return tryGetLock(key, value, expire, TimeUnit.SECONDS);
}
public Boolean tryGetLock(String key , String value, Integer expire , TimeUnit timeUnit) {
    ValueOperations operations = redisTemplate.opsForValue();
    if (operations.setIfAbsent(key, value)) {
        //說(shuō)明 redis沒(méi)有該key , 換言之 加鎖成功  設置過(guò)期時(shí)間防止死鎖
        if (expire > 0) {
            redisTemplate.expire(key, expire, timeUnit);
        }
        return true;
    }
    return false;
}

public Boolean unLock(String key) {
    return redisTemplate.delete(key);
}

StockKeyGenerator

@Component()
@Primary
public class StockKeyGenerator implements CacheKeyGenerator {
    @Override
    public String getLockKey(ProceedingJoinPoint pjp) {
        //獲取方法簽名
        MethodSignature signature = (MethodSignature) pjp.getSignature();
        Method method = signature.getMethod();
        //獲取方法cacheLock注解
        StockLock stockLock = method.getAnnotation(StockLock.class);
        //獲取方法參數
        Object[] args = pjp.getArgs();
        Parameter[] parameters = method.getParameters();
        StringBuilder builder = new StringBuilder();
        for (int i = 0; i < parameters.length; i++) {
            StockParam stockParam = parameters[i].getAnnotation(StockParam.class);
            Object arg = args[i];
            if (arg instanceof Map) {
                Map<String, Object> temArgMap = (Map<String, Object>) arg;
                String[] names = stockParam.names();
                for (String name : names) {
                    if (builder.length() > 0) {
                        builder.append(stockLock.delimiter());
                    }
                    builder.append(temArgMap.get(name));
                }
            }

        }
        return builder.toString();
    }
}

問(wèn)題分析

上面分析了一個(gè)死鎖的場(chǎng)景，理論上出了死鎖我們r(jià)edis分布鎖很好的解決了分布式問(wèn)題。但是還是會(huì )出現問(wèn)題。下面列舉寫(xiě)小編遇到的問(wèn)題。

業(yè)務(wù)處理時(shí)間>上鎖過(guò)期時(shí)間

• a線(xiàn)程獲取到鎖，開(kāi)始進(jìn)行業(yè)務(wù)處理需要8S，
• 在8S內，鎖的有效期是5S，在鎖過(guò)期后也就是第6S ， b線(xiàn)程進(jìn)入開(kāi)始獲取鎖這個(gè)時(shí)候b是可以獲取到新鎖的。這個(gè)時(shí)候就是有問(wèn)題的。
• 假設b線(xiàn)程業(yè)務(wù)處理只需要3S ， 但是因為a線(xiàn)程釋放了鎖，所以在第8S的時(shí)候雖然b線(xiàn)程沒(méi)有釋放鎖，b的鎖也沒(méi)有過(guò)期但是這時(shí)候也沒(méi)有了鎖。從而導致C線(xiàn)程也可以進(jìn)入

以上就是Java基于redis實(shí)現分布式鎖的詳細內容，更多關(guān)于Java 實(shí)現分布式鎖的資料請關(guān)注腳本之家其它相關(guān)文章！