【Java項(xiàng)目】高并發(fā)場(chǎng)景下，基于Redisson實(shí)現(xiàn)的分布式鎖

分布式鎖應(yīng)用場(chǎng)景

隨著互聯(lián)網(wǎng)應(yīng)用的高速發(fā)展，在電商應(yīng)用中高并發(fā)應(yīng)用場(chǎng)景涉及很多，例如：

• 秒殺:在大規(guī)模的秒殺場(chǎng)景中，需要保證商品數(shù)量、限制用戶(hù)購(gòu)買(mǎi)數(shù)量, 防止用戶(hù)購(gòu)買(mǎi)數(shù)量的超限、避免出現(xiàn)超賣(mài)情況;
• 訂單支付: 當(dāng)用戶(hù)下單付款時(shí)，需要對(duì)訂單信息進(jìn)行互斥操作以避免訂單重復(fù)支付;
• 提現(xiàn)操作：需要防止用戶(hù)重復(fù)提現(xiàn)，避免造成財(cái)務(wù)損失。
• 總結(jié):分布式鎖應(yīng)用場(chǎng)景可以分為兩類(lèi)：
• 1、共享資源的互斥訪(fǎng)問(wèn)：當(dāng)多個(gè)節(jié)點(diǎn)需要對(duì)同一個(gè)共享資源進(jìn)行操作時(shí)，需要確保同一時(shí)刻只有一個(gè)節(jié)點(diǎn)可以操作，此時(shí)就可以使用分布式鎖；
• 2、分布式任務(wù)調(diào)度：分布式系統(tǒng)往往需要對(duì)任務(wù)進(jìn)行調(diào)度，確保任務(wù)在多個(gè)節(jié)點(diǎn)的協(xié)作下執(zhí)行。而在并行的任務(wù)執(zhí)行過(guò)程中，需要區(qū)分哪些任務(wù)已經(jīng)被分配并且正在被執(zhí)行，哪些任務(wù)沒(méi)有被分配。利用分布式鎖來(lái)保證任務(wù)的正確性、順序性和穩(wěn)定性。
• 概括地說(shuō)，就是對(duì)多線(xiàn)程下，對(duì)共享變量操作，線(xiàn)程間是變量不可見(jiàn)，導(dǎo)致出現(xiàn)并發(fā)問(wèn)題，需要通過(guò)分布式鎖來(lái)進(jìn)行控制，今天就給大家通過(guò)案例，分享一下如何使用redisson實(shí)現(xiàn)分布式鎖。

案例需求描述

庫(kù)存中有200件商品，通過(guò)商品下單購(gòu)買(mǎi)場(chǎng)景，使用分布式鎖避免商品超賣(mài)問(wèn)題。

Redisson環(huán)境準(zhǔn)備

本地Redis環(huán)境安裝

下載地址： https://github.com/tporadowski/redis/releases

1、windows下安裝

默認(rèn)端口：6379

下載連接 https://github.com/tporadowski/redis/releases

解壓

雙擊redis-server.exe啟動(dòng)服務(wù)端

雙擊redis-cli.exe啟動(dòng)客戶(hù)端連接服務(wù)端

在客戶(hù)端輸入 “ping”，出現(xiàn)“PONG”，即證明連接成功,部分配置可以在redis.conf文件修改；

Spring boot項(xiàng)目與redis集成

引入依賴(lài)

?<dependency>
? <groupId>org.redisson</groupId>
? <artifactId>redisson-spring-boot-starter</artifactId>
? <version>3.22.0</version>
?</dependency>

創(chuàng)建redis連接池代碼

?package com.zhc.config.redis;
?
?import org.redisson.Redisson;
?import org.redisson.api.RedissonClient;
?import org.redisson.client.codec.Codec;
?import org.redisson.codec.JsonJacksonCodec;
?import org.redisson.config.Config;
?import org.springframework.beans.factory.annotation.Value;
?import org.springframework.context.annotation.Bean;
?import org.springframework.context.annotation.Configuration;
?import org.springframework.data.redis.connection.RedisConnectionFactory;
?import org.springframework.data.redis.core.RedisTemplate;
?import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
?import org.springframework.data.redis.serializer.StringRedisSerializer;
?
?/**
? * redisson 連接池配置
? * @author zhouhengchao
? * @since 2023-06-19 20:29:00
? * @version 1.0
? */
?@Configuration
?public class RedisConfig {
?
??@Value("${spring.redis.host}")
??private String host;
?
??@Value("${spring.redis.port}")
??private String port;
?
??@Value("${spring.redis.database}")
??private Integer dataBase;
?
??@Bean(name = "redisTemplate")
??public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
????GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = serializer();
????RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
????// key采用String的序列化方式
????redisTemplate.setKeySerializer(StringRedisSerializer.UTF_8);
????// value序列化方式采用jackson
????redisTemplate.setValueSerializer(genericJackson2JsonRedisSerializer);
????// hash的key也采用String的序列化方式
????redisTemplate.setHashKeySerializer(StringRedisSerializer.UTF_8);
????//hash的value序列化方式采用jackson
????redisTemplate.setHashValueSerializer(genericJackson2JsonRedisSerializer);
????redisTemplate.setConnectionFactory(redisConnectionFactory);
????return redisTemplate;
??}
?
??/**
???* 此方法不能用@Ben注解，避免替換Spring容器中的同類(lèi)型對(duì)象
???*/
??public GenericJackson2JsonRedisSerializer serializer() {
????return new GenericJackson2JsonRedisSerializer();
??}
?
??@Bean
??public RedissonClient redissonClient() {
????Config config = new Config();
????config.useSingleServer().setAddress("redis://" + host + ":" + port).setDatabase(dataBase);
????// 設(shè)置redisson序列化方式，否則打開(kāi)查看數(shù)據(jù)可能亂碼
????Codec codec = new JsonJacksonCodec();
????config.setCodec(codec);
????return Redisson.create(config);
??}
?}

redis的yaml文件配置

?spring:
? redis:
??host: localhost
??port: 6379
??database: 0

扣減庫(kù)存方法

?/**
???* 從redis中獲取庫(kù)存，扣減庫(kù)存數(shù)量
???*/
??private void reduceStock(){
????// 從redis中獲取商品庫(kù)存
????RBucket<Integer> bucket = redissonClient.getBucket(REDIS_STOCK);
????int stock = bucket.get();
????if (stock > 0) {
??????// 庫(kù)存-1
??????stock--;
??????// 更新庫(kù)存
??????bucket.set(stock, 2, TimeUnit.DAYS);
??????log.info("扣減成功，剩余庫(kù)存：" + stock);
????} else {
??????log.info("扣減失敗，庫(kù)存不足");
????}
??}

基于synchronized加鎖控制

?@GetMapping("/test01")
??public void test01(){
????for (int i = 0; i < 6; i++) {
??????synchronized (this) {
????????new Thread(this::reduceStock).start();
??????}
????}
??}
???

我們通過(guò)了Synchronized鎖，成功解決了多個(gè)線(xiàn)程爭(zhēng)搶導(dǎo)致的超賣(mài)問(wèn)題，但是有個(gè)問(wèn)題，假設(shè)后期公司為了保證服務(wù)可用性。

將單擊的應(yīng)用，升級(jí)稱(chēng)為了集群的模式，那么是否會(huì)有超賣(mài)問(wèn)題呢？

通過(guò)nginx搭建負(fù)載均衡

下載Nginx：?http://nginx.org/download/nginx-1.18.0.zip

nginx.conf完整配置

?worker_processes 1;
?events {
??worker_connections 1024;
?}
?http {
??include???mime.types;
??default_type application/octet-stream;
??sendfile???on;
??keepalive_timeout 65;
? upstream redislock{
??server localhost:8081 weight=1;
??server localhost:8082 weight=1;
?}
??server {
????listen???80;
????server_name localhost;
????location / {
??????root?html;
??????index index.html index.htm;
?   proxy_pass http://redislock;
????}
????error_page?500 502 503 504 /50x.html;
????location = /50x.html {
??????root?html;
????}
??}
?
?}
?

啟動(dòng)nginx，雙擊nginx.exe文件即可；

訪(fǎng)問(wèn)應(yīng)用：http://localhost/test01

發(fā)現(xiàn)存在超賣(mài)問(wèn)題。

使用redis分布式鎖

?@GetMapping("/test02")
??public void test02(){
????// 分布式鎖名稱(chēng),關(guān)鍵是多個(gè)應(yīng)用要共享這一個(gè)Redis的key
????String lockKey = "lockDeductStock";
????// setIfAbsent 如果不存在key就set值，并返回1
????//如果存在(不為空)不進(jìn)行操作，并返回0,與redis命令setnx相似，setIfAbsent是java中的方法
????// 根據(jù)返回值為1就表示獲取分布式鎖成功，返回0就表示獲取鎖失敗
????Boolean lockResult = redisTemplate.opsForValue().setIfAbsent(lockKey, lockKey);
????// 加鎖不成功，返回給前端錯(cuò)誤碼，前端給用戶(hù)友好提示
????if (Boolean.FALSE.equals(lockResult)) {
??????log.info("系統(tǒng)繁忙，請(qǐng)稍后再試！");
??????return;
????}
????reduceStock();
????// 業(yè)務(wù)執(zhí)行完成，刪除這個(gè)鎖
????redisTemplate.delete(lockKey);
??}

1、主要使用setIfAbsent方法：如果不包含key就set值，并返回1；

如果存在(不為空)不進(jìn)行操作，并返回0;

2、很明顯，比get和set要好。因?yàn)橄扰袛鄃et，再set的用法，有可能會(huì)重復(fù)set值，與setnx類(lèi)似。

以上redis加鎖可以解決并發(fā)問(wèn)題，但是存在問(wèn)題：

1、如果setIfAbsent加鎖成功，但是到業(yè)務(wù)邏輯代碼時(shí)，該服務(wù)掛掉了，就會(huì)導(dǎo)致另一個(gè)服務(wù)一直獲取不到鎖，一直在等待中；

2、可以使用 redisTemplate.opsForValue().setIfAbsent(lockKey, lockKey,30,TimeUnit.SECONDS)，設(shè)置鎖的key過(guò)期時(shí)間，在規(guī)定時(shí)間后key過(guò)期就可以再獲取。

redis分布式鎖優(yōu)化

以上分布式鎖還是存在問(wèn)題，如果鎖的key過(guò)期時(shí)間與程序執(zhí)行時(shí)間差問(wèn)題，例如

• 如果鎖key在程序執(zhí)行結(jié)束前過(guò)期，就會(huì)導(dǎo)致刪除key失敗；
• 同時(shí)另一個(gè)應(yīng)用獲取了鎖，又會(huì)被其他應(yīng)用刪掉鎖，導(dǎo)致鎖一直失效，存在并發(fā)問(wèn)題。
• 可以通過(guò)引入U(xiǎn)UId來(lái)解決鎖被其他應(yīng)用勿釋放問(wèn)題，如下代碼：

?@GetMapping("/test03")
??public void test03(){
????// 分布式鎖名稱(chēng),關(guān)鍵是多個(gè)應(yīng)用要共享這一個(gè)Redis的key
????String lockKey = "lockDeductStock";
????// 分布式鎖的值
????String lockValue = UUID.randomUUID().toString().replaceAll("-",   "");
????// setIfAbsent 如果不存在key就set值，并返回1
????//如果存在(不為空)不進(jìn)行操作，并返回0,與redis命令setnx相似，setIfAbsent是java中的方法
????// 根據(jù)返回值為1就表示獲取分布式鎖成功，返回0就表示獲取鎖失敗
????Boolean lockResult =   redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 30, TimeUnit.SECONDS);
????// 加鎖不成功，返回給前端錯(cuò)誤碼，前端給用戶(hù)友好提示
????if (Boolean.FALSE.equals(lockResult)) {
??????log.info("系統(tǒng)繁忙，請(qǐng)稍后再試！");
??????return ;
????}
????reduceStock();
????// 判斷是不是當(dāng)前請(qǐng)求的UUID，如果是則可以正常釋放鎖。如果不是，則釋放鎖失敗！
????if (lockValue.equals(redisTemplate.opsForValue().get(lockKey)))   {
??????redisTemplate.delete(lockKey);
????}
??}

還存在鎖超時(shí)問(wèn)題：鎖超時(shí)問(wèn)題，寫(xiě)一個(gè)定時(shí)任務(wù)，分線(xiàn)程每隔十秒去查看一次主線(xiàn)程是否持有這把鎖，如果這個(gè)鎖存在，重新將這個(gè)鎖的超時(shí)時(shí)間設(shè)置為30S，對(duì)鎖延時(shí)，比較復(fù)雜。

使用redisson實(shí)現(xiàn)分布式鎖

?@GetMapping("/test04")
??public void test04(){
????// 分布式鎖名稱(chēng),關(guān)鍵是多個(gè)應(yīng)用要共享這一個(gè)Redis的key
????String lockKey = "lockDeductStock";
????// 獲取鎖對(duì)象
????RLock redissonLock = redissonClient.getLock(lockKey);
????try {
??????redissonLock.lock();
?//?????boolean result = redissonLock.tryLock();
??????// 加鎖不成功，返回給前端錯(cuò)誤碼，前端給用戶(hù)友好提示
?//?????if (!result) {
?//???????log.info("系統(tǒng)繁忙，請(qǐng)稍后再試！");
?//???????return;
?//?????}
??????reduceStock();
????}
????finally{
??????if(redissonLock.isHeldByCurrentThread()){
????????redissonLock.unlock();
??????}
????}
??}

redisson分布式鎖原理圖：

關(guān)鍵方法介紹：

• lock() 方法是阻塞獲取鎖的方式，如果當(dāng)前鎖被其他線(xiàn)程持有，則當(dāng)前線(xiàn)程會(huì)一直阻塞等待獲取鎖，直到獲取到鎖或者發(fā)生超時(shí)或中斷等情況才會(huì)結(jié)束等待;
• tryLock() 方法是一種非阻塞獲取鎖的方式，在嘗試獲取鎖時(shí)不會(huì)阻塞當(dāng)前線(xiàn)程，而是立即返回獲取鎖的結(jié)果，如果獲取成功則返回 true，否則返回 false.
• 總結(jié)：
• lock()方法獲取到鎖之后可以保證線(xiàn)程對(duì)共享資源的訪(fǎng)問(wèn)是互斥的，適用于需要確保共享資源只能被一個(gè)線(xiàn)程訪(fǎng)問(wèn)的場(chǎng)景。Redisson 的 lock() 方法支持可重入鎖和公平鎖等特性，可以更好地滿(mǎn)足多線(xiàn)程并發(fā)訪(fǎng)問(wèn)的需求；
• tryLock() 方法支持加鎖時(shí)間限制、等待時(shí)間限制以及可重入等特性，可以更好地控制獲取鎖的過(guò)程和等待時(shí)間，避免程序出現(xiàn)長(zhǎng)時(shí)間無(wú)法響應(yīng)等問(wèn)題。
• 在實(shí)際應(yīng)用中需要根據(jù)具體場(chǎng)景和業(yè)務(wù)需求來(lái)選擇合適的方法，以確保程序的正確性和高效性。
• 視頻中的內(nèi)容如果對(duì)您有所幫助，請(qǐng)給個(gè)三連加關(guān)注的支持，歡迎在評(píng)論區(qū)留言討論，后續(xù)會(huì)進(jìn)一步完善文檔。