Spring boot集成Redis

一：redis基本集成

首先對redis來說，所有的key（鍵）都是字符串。

我們在談基礎數據結構時，討論的是存儲值的數據類型，主要包括常見的5種數據類型，分別是：String、List、Set、Zset、Hash。

結構類型	結構存儲的值	結構的讀寫能力
String	可以是字符串、整數或浮點數	對整個字符串或字符串的一部分進行操作；對整數或浮點數進行自增或自減操作；
List	一個鏈表，鏈表上的每個節點都包含一個字符串	對鏈表的兩端進行push和pop操作，讀取單個或多個元素；根據值查找或刪除元素；
Hash	包含鍵值對的無序散列表	包含方法有添加、獲取、刪除單個元素
Set	包含字符串的無序集合	字符串的集合，包含基礎的方法有看是否存在添加、獲取、刪除； 還包含計算交集、并集、差集等
Zset	和散列一樣，用于存儲鍵值對	字符串成員與浮點數分數之間的有序映射； 元素的排列順序由分數的大小決定； 包含方法有添加、獲取、刪除單個元素以及根據分值范圍或成員來獲取元素

1：RedisTemplate + Jedis

Jedis是Redis的Java客戶端，在SpringBoot 1.x版本中也是默認的客戶端。

在SpringBoot 2.x版本中默認客戶端是Luttuce。

1.1：RedisTemplate

Spring 通過模板方式（RedisTemplate）提供了對Redis的數據查詢和操作功能。

什么是模板方法模式

模板方法模式(Template pattern): 在一個方法中定義一個算法的骨架, 而將一些步驟延遲到子類中.

模板方法使得子類可以在不改變算法結構的情況下, 重新定義算法中的某些步驟。

RedisTemplate對于Redis5種基礎類型的操作

redisTemplate.opsForValue(); // 操作字符串
redisTemplate.opsForHash(); // 操作hash
redisTemplate.opsForList(); // 操作list
redisTemplate.opsForSet(); // 操作set
redisTemplate.opsForZSet(); // 操作zset

對HyperLogLogs（基數統計）類型的操作

redisTemplate.opsForHyperLogLog();

對geospatial (地理位置)類型的操作

redisTemplate.opsForGeo();

對于BitMap的操作，也是在opsForValue()方法返回類型ValueOperations中

Boolean setBit(K key, long offset, boolean value);
Boolean getBit(K key, long offset);

對于Stream的操作

redisTemplate.opsForStream();

1.2：實現案例

本例子主要基于SpringBoot2+ 使用Jedis客戶端，通過RedisTemplate模板方式訪問Redis數據。

其他實體類結構請看（集成Jpa）

1.2.1：依賴引入和屬性配置

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><!-- 暫時先排除lettuce-core，使用jedis --><!-- jedis是spring-boot 1.x的默認，lettuce是spring-boot 2.x的默認 --><exclusions><exclusion><artifactId>lettuce-core</artifactId><groupId>io.lettuce</groupId></exclusion></exclusions>
</dependency><!-- 格外使用jedis -->
<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId>
</dependency><!-- commons-pools，連接池 -->
<dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId><version>2.9.0</version>
</dependency>

spring:# swagger配置mvc:path match:# 由于 springfox 3.0.x 版本 和 Spring Boot 2.6.x 版本有沖突，所以還需要先解決這個 bugmatching-strategy: ANT_PATH_MATCHER# 數據源配置datasource:url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghaidriver-class-name: com.mysql.cj.jdbc.Driver # 8.0 +username: rootpassword: bnm314159# JPA 配置jpa:generate-ddl: false # 是否自動創建數據庫表show-sql: true # 是否打印生成的 sqlproperties:hibernate:dialect: org.hibernate.dialect.MySQL8Dialect # 數據庫方言 mysql8format_sql: true # 是否格式化 sqluse_new_id_generator_mappings: true # 是否使用新的 id 生成器# redis 配置redis:database: 0 # redis數據庫索引（默認為0）host: 127.0.0.1 # redis服務器地址port: 6379 # redis服務器連接端口jedis:pool:min-idle: 0 # 連接池中的最小空閑連接max-active: 8 # 連接池最大連接數（使用負值表示沒有限制）max-idle: 8 # 連接池中的最大空閑連接max-wait: -1ms # 連接池最大阻塞等待時間（使用負值表示沒有限制）connect-timeout: 30000ms # 連接超時時間（毫秒）timeout: 30000ms # 讀取超時時間（毫秒）

1.2.2：redisConfig配置

package com.cui.jpa_demo.config;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;/*** @author cui haida* 2025/1/25*/
@Configuration
public class RedisConfig {/*** 配置RedisTemplate以支持鍵值對存儲* 該方法在Spring框架中定義了一個Bean，用于創建和配置RedisTemplate實例* RedisTemplate用于與Redis數據庫進行交互，支持數據的存儲和檢索** @param factory RedisConnectionFactory實例，用于連接Redis服務器* @return 配置好的RedisTemplate實例，用于執行鍵值對操作*/@Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {// 創建RedisTemplate實例，并指定鍵和值的類型RedisTemplate<String, Object> template = new RedisTemplate<>();// 設置連接工廠，用于建立與Redis服務器的連接template.setConnectionFactory(factory);// 配置鍵的序列化方式為StringRedisSerializer// 這是為了確保鍵以字符串形式存儲和檢索template.setKeySerializer(new StringRedisSerializer());// 配置哈希鍵的序列化方式為StringRedisSerializer// 這適用于哈希表中的鍵值對操作template.setHashKeySerializer(new StringRedisSerializer());// 配置值的序列化方式為GenericJackson2JsonRedisSerializer// 使用Jackson庫將對象序列化為JSON格式存儲template.setValueSerializer(new GenericJackson2JsonRedisSerializer());// 配置哈希表值的序列化方式為GenericJackson2JsonRedisSerializer// 同樣使用Jackson庫將對象序列化為JSON格式存儲template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());// 初始化RedisTemplate，確保所有必需的屬性都已設置template.afterPropertiesSet();// 返回配置好的RedisTemplate實例return template;}
}

1.2.3：基礎使用

package com.cui.jpa_demo.controller;import com.cui.jpa_demo.entity.bean.UserQueryBean;
import com.cui.jpa_demo.entity.model.User;
import com.cui.jpa_demo.entity.response.ResponseResult;
import com.cui.jpa_demo.service.IUserService;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;
import java.time.LocalDateTime;/*** @author cui haida* 2025/1/23*/
@RestController
@RequestMapping("/user")
public class UserController {private final IUserService userService;@Resourceprivate RedisTemplate<String, User> redisTemplate;public UserController(IUserService userService) {this.userService = userService;}@PostMapping("add")public ResponseResult<User> add(User user) {if (user.getId()==null || !userService.exists(user.getId())) {user.setCreateTime(LocalDateTime.now());user.setUpdateTime(LocalDateTime.now());userService.save(user);} else {user.setUpdateTime(LocalDateTime.now());userService.update(user);}return ResponseResult.success(userService.find(user.getId()));}/*** @return user list*/@GetMapping("edit/{userId}")public ResponseResult<User> edit(@PathVariable("userId") Long userId) {return ResponseResult.success(userService.find(userId));}/*** @return user list*/@GetMapping("list")public ResponseResult<Page<User>> list(@RequestParam int pageSize, @RequestParam int pageNumber) {return ResponseResult.success(userService.findPage(UserQueryBean.builder().build(), PageRequest.of(pageNumber, pageSize)));}@PostMapping("/redis/add")public ResponseResult<User> addIntoRedis(User user) {redisTemplate.opsForValue().set(String.valueOf(user.getId()), user);return ResponseResult.success(redisTemplate.opsForValue().get(String.valueOf(user.getId())));}@GetMapping("/redis/get/{userId}")public ResponseResult<User> getFromRedis(@PathVariable("userId") Long userId) {return ResponseResult.success(redisTemplate.opsForValue().get(String.valueOf(userId)));}
}

2：RedisTemplate+Lettuce

2.1：什么是Lettuce

Lettuce 是一個可伸縮線程安全的 Redis 客戶端。多個線程可以共享同一個 RedisConnection。

它利用優秀 netty NIO 框架來高效地管理多個連接。

Lettuce的特性：

• 支持 同步、異步、響應式 的方式
• 支持 Redis Sentinel
• 支持 Redis Cluster
• 支持 SSL 和 Unix Domain Socket 連接
• 支持 Streaming API
• 支持 CDI 和 Spring 的集成
• 支持 Command Interfaces
• 兼容 Java 8+ 以上版本

2.2：為何能干掉Jedis成為默認

除了上述特性的支持性之外，最為重要的是Lettuce中使用了Netty框架，使其具備線程共享和異步的支持性。

線程共享

Jedis 是直連模式，在多個線程間共享一個 Jedis 實例時是線程不安全的

如果想要在多線程環境下使用 Jedis，需要使用連接池，每個線程都去拿自己的 Jedis 實例，當連接數量增多時，物理連接成本就較高了。

Lettuce 是基于 netty 的，連接實例可以在多個線程間共享，所以，一個多線程的應用可以使用一個連接實例，而不用擔心并發線程的數量。

異步和反應式

Lettuce 從一開始就按照非阻塞式 IO 進行設計，是一個純異步客戶端，對異步和反應式 API 的支持都很全面。

即使是同步命令，底層的通信過程仍然是異步模型，只是通過阻塞調用線程來模擬出同步效果而已。

2.3：Lettuce的基本的API方式

依賴POM包

<dependency><groupId>io.lettuce</groupId><artifactId>lettuce-core</artifactId><version>x.y.z.BUILD-SNAPSHOT</version>
</dependency>

基礎用法

// 聲明redis-client
RedisClient client = RedisClient.create("redis://localhost");
// 創建連接
StatefulRedisConnection<String, String> connection = client.connect();
// 同步命令
RedisStringCommands sync = connection.sync();
// 執行get方法
String value = sync.get("key");

異步方式

StatefulRedisConnection<String, String> connection = client.connect();
// 異步命令
RedisStringAsyncCommands<String, String> async = connection.async();
// 異步set & get
RedisFuture<String> set = async.set("key", "value")
RedisFuture<String> get = async.get("key")async.awaitAll(set, get) == trueset.get() == "OK"
get.get() == "value"

• 響應式

StatefulRedisConnection<String, String> connection = client.connect();
RedisStringReactiveCommands<String, String> reactive = connection.reactive();
Mono<String> set = reactive.set("key", "value");
Mono<String> get = reactive.get("key");
// 訂閱
set.subscribe();get.block() == "value"

2.4：實現案例

依賴和配置

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId>
</dependency><!-- 一定要加入這個，否則連接池用不了 --> 
<dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId>
</dependency>

配置

  # redis 配置redis:host: 127.0.0.1 # 地址port: 6379 # 端口database: 0 # redis 數據庫索引# 如果是集群模式，需要配置如下
#    cluster:
#      nodes:
#        - 127.0.0.1:7000
#        - 127.0.0.1:7001
#        - 127.0.0.1:7002lettuce:pool:max-wait: -1 # 最大連接等待時間, 默認 -1 表示沒有限制max-active: 8 # 最大連接數, 默認8max-idle: 8 # 最大空閑連接數, 默認8min-idle: 0 # 最小空閑連接數, 默認0
#    password: 123456 # 密碼
#    timeout: 10000ms # 超時時間
#    ssl: false # 是否啟用 SSL
#    sentinel:
#      master: mymaster # 主節點名稱
#      nodes: 127.0.0.1:26379,127.0.0.1:26380,127.0.0.1:26381 # 哨兵節點

序列化配置

redis的序列化也是我們在使用RedisTemplate的過程中需要注意的事情。

如果沒有特殊設置redis的序列化方式，那么它其實使用的是默認的序列化方式【JdkSerializationRedisSerializer】。

這種序列化最大的問題就是存入對象后，我們很難直觀看到存儲的內容，很不方便我們排查問題

RedisTemplate這個類的泛型是<String,Object>, 也就是他是支持寫入Object對象的，那么這個對象采取什么方式序列化存入內存中就是它的序列化方式。

Redis本身提供了以下幾種序列化的方式：

• GenericToStringSerializer: 可以將任何對象泛化為字符串并序列化
• Jackson2JsonRedisSerializer: 跟JacksonJsonRedisSerializer實際上是一樣的 <---- 我們要換成這個
• JacksonJsonRedisSerializer: 序列化object對象為json字符串
• JdkSerializationRedisSerializer: 序列化java對象【默認的】
• StringRedisSerializer: 簡單的字符串序列化 JSON 方式序列化成字符串，存儲到 Redis 中 。我們查看的時候比較直觀

package com.study.study_demo_of_spring_boot.redis_study.config;import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.jsontype.impl.LaissezFaireSubTypeValidator;
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.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;/*** <p>* 功能描述：redis 序列化配置類* </p>** @author cui haida* @date 2024/04/13/19:52*/
@Configuration
public class RedisConfig {/*** 創建并配置RedisTemplate，用于操作Redis數據庫。** @param factory Redis連接工廠，用于創建Redis連接。* @return 配置好的RedisTemplate對象，可以用于執行Redis操作。*/@Bean(name = "redisTemplate")public RedisTemplate<String, Object> getRedisTemplate(RedisConnectionFactory factory) {RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();redisTemplate.setConnectionFactory(factory);// 配置Key的序列化方式為StringRedisSerializerStringRedisSerializer stringRedisSerializer = new StringRedisSerializer();redisTemplate.setKeySerializer(stringRedisSerializer);// 配置Value的序列化方式為Jackson2JsonRedisSerializerJackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);ObjectMapper objectMapper = new ObjectMapper();objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);jackson2JsonRedisSerializer.setObjectMapper(objectMapper);redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);// 配置Hash的Key和Value的序列化方式redisTemplate.setHashKeySerializer(stringRedisSerializer);redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);// 初始化RedisTemplateredisTemplate.afterPropertiesSet();return redisTemplate;}
}

業務類調用

import io.swagger.annotations.ApiOperation;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.*;
import tech.pdai.springboot.redis.lettuce.entity.User;
import tech.pdai.springboot.redis.lettuce.entity.response.ResponseResult;import javax.annotation.Resource;@RestController
@RequestMapping("/user")
public class UserController {// 注意：這里@Autowired是報錯的，因為@Autowired按照類名注入的@Resourceprivate RedisTemplate<String, User> redisTemplate;/*** @param user user param* @return user*/@ApiOperation("Add")@PostMapping("add")public ResponseResult<User> add(User user) {redisTemplate.opsForValue().set(String.valueOf(user.getId()), user);return ResponseResult.success(redisTemplate.opsForValue().get(String.valueOf(user.getId())));}/*** @return user list*/@ApiOperation("Find")@GetMapping("find/{userId}")public ResponseResult<User> edit(@PathVariable("userId") String userId) {return ResponseResult.success(redisTemplate.opsForValue().get(userId));}
}

2.5：數據類封裝

RedisTemplate中的操作和方法眾多，為了程序保持方法使用的一致性，屏蔽一些無關的方法以及對使用的方法進一步封裝。

import org.springframework.data.redis.core.RedisCallback;import java.util.Collection;
import java.util.Set;/*** 可能只關注這些方法*/
public interface IRedisService<T> {void set(String key, T value);void set(String key, T value, long time);T get(String key);void delete(String key);void delete(Collection<String> keys);boolean expire(String key, long time);Long getExpire(String key);boolean hasKey(String key);Long increment(String key, long delta);Long decrement(String key, long delta);void addSet(String key, T value);Set<T> getSet(String key);void deleteSet(String key, T value);T execute(RedisCallback<T> redisCallback);
}

RedisService的實現類

import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import tech.pdai.springboot.redis.lettuce.enclosure.service.IRedisService;import javax.annotation.Resource;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.TimeUnit;@Service
public class RedisServiceImpl<T> implements IRedisService<T> {@Resourceprivate RedisTemplate<String, T> redisTemplate;@Overridepublic void set(String key, T value, long time) {redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);}@Overridepublic void set(String key, T value) {redisTemplate.opsForValue().set(key, value);}@Overridepublic T get(String key) {return redisTemplate.opsForValue().get(key);}@Overridepublic void delete(String key) {redisTemplate.delete(key);}@Overridepublic void delete(Collection<String> keys) {redisTemplate.delete(keys);}@Overridepublic boolean expire(String key, long time) {return redisTemplate.expire(key, time, TimeUnit.SECONDS);}@Overridepublic Long getExpire(String key) {return redisTemplate.getExpire(key, TimeUnit.SECONDS);}@Overridepublic boolean hasKey(String key) {return redisTemplate.hasKey(key);}@Overridepublic Long increment(String key, long delta) {return redisTemplate.opsForValue().increment(key, delta);}@Overridepublic Long decrement(String key, long delta) {return redisTemplate.opsForValue().increment(key, -delta);}@Overridepublic void addSet(String key, T value) {redisTemplate.opsForSet().add(key, value);}@Overridepublic Set<T> getSet(String key) {return redisTemplate.opsForSet().members(key);}@Overridepublic void deleteSet(String key, T value) {redisTemplate.opsForSet().remove(key, value);}@Overridepublic T execute(RedisCallback<T> redisCallback) {return redisTemplate.execute(redisCallback);}
}

RedisService的調用

@RestController
@RequestMapping("/user")
public class UserController {@Autowiredprivate IRedisService<User> redisService;//...
}

二：集成redisson

1：單機可重入鎖

redisson-spring-boot-starter依賴于與最新版本的spring-boot兼容的redisson-spring數據模塊。

redisson-spring-data module name	spring boot version
redisson-spring-data-16	1.3.y
redisson-spring-data-17	1.4.y
redisson-spring-data-18	1.5.y
redisson-spring-data-2x	2.x.y
redisson-spring-data-3x	3.x.y

<!-- redisson -->
<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.16.2</version>
</dependency>

package com.study.study_demo_of_spring_boot.redis_study.config;import lombok.Data;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.io.IOException;/*** <p>* 功能描述：redis client 配置* </p>** @author cui haida* @date 2024/04/14/7:24*/
@Configuration
@ConfigurationProperties(prefix = "spring.redis")
@Data
public class MyRedissonConfig {private String host;private int port;@Bean(destroyMethod = "shutdown")RedissonClient redisson() throws IOException {Config config = new Config();config.useSingleServer().setAddress("redis://" + host + ":" + port);return Redisson.create(config);}
}

加鎖解鎖測試

package com.study.study_demo_of_spring_boot.redis_study.use;import com.study.study_demo_of_spring_boot.redis_study.config.MyRedissonConfig;
import com.study.study_demo_of_spring_boot.redis_study.util.RedisUtil;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.util.concurrent.TimeUnit;/*** <p>* 功能描述：redis test* </p>** @author cui haida* @date 2024/04/14/7:28*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class UseTest {@Autowiredprivate RedisUtil redisUtil;@Autowiredprivate RedissonClient redissonClient;@Testpublic void redisNormalTest() {redisUtil.set("name", "張三");}@Testpublic void redissonTest() {RLock lock = redissonClient.getLock("global_lock_key");try {System.out.println(lock);// 加鎖30mslock.lock(30, TimeUnit.MILLISECONDS);if (lock.isLocked()) {System.out.println("獲取到了");} else {System.out.println("未獲取到");}} catch (Exception e) {e.printStackTrace();} finally {if (lock.isHeldByCurrentThread()) {lock.unlock();System.out.println("解鎖成功");}}}
}

• lock.lock()即沒有指定鎖的過期時間，就是用30s，即看門狗的默認時間，只要占鎖成功，就會啟動一個定時任務，每隔10秒就會自動續期到30秒。
• lock.lock(10, TimeUnit.xxx)，默認鎖的過期時間就是我們指定的時間。

2：紅鎖（Red Lock）

紅鎖其實就是對多個redission節點同時加鎖

2.1：環境準備

用docker啟動三個redis實例，模擬redLock

docker run -itd  # -d 后臺啟動, -it shell交互--name redlock-1  # 這個容器的名稱-p 6380:6379 # 端口映射 redis的6379 <-> 容器的6380映射redis:7.0.8 # 鏡像名稱，如果沒有下載對應的redis鏡像，將會先進行拉取--requirepass 123456 # redis密碼123456
docker run -itd --name redlock-2 -p 6381:6379 redis:7.0.8 --requirepass 123456
docker run -itd --name redlock-3 -p 6382:6379 redis:7.0.8 --requirepass 123456

2.2：配置編寫

package com.study.study_demo_of_spring_boot.redis_study.config;import lombok.Data;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.io.IOException;/*** <p>* 功能描述：redis client 配置* </p>** @author cui haida* @date 2024/04/14/7:24*/
@Configuration
@ConfigurationProperties(prefix = "spring.redis")
@Data
public class MyRedissonConfig {private String host;private int port;@Bean(name = "normalRedisson", destroyMethod = "shutdown")RedissonClient redisson() {Config config = new Config();config.useSingleServer().setAddress("redis://" + host + ":" + port);return Redisson.create(config);}@Bean(name = "redLock1")RedissonClient redissonClient1(){Config config = new Config();config.useSingleServer().setAddress("redis://ip:6380").setDatabase(0).setPassword("123456");return Redisson.create(config);}@Bean(name = "redLock2")RedissonClient redissonClient2(){Config config = new Config();config.useSingleServer().setAddress("redis://ip:6381").setDatabase(0).setPassword("123456");return Redisson.create(config);}@Bean(name = "redLock3")RedissonClient redissonClient3(){Config config = new Config();config.useSingleServer().setAddress("redis://ip:6382").setDatabase(0).setPassword("123456");return Redisson.create(config);}
}

2.3：使用測試

package com.study.study_demo_of_spring_boot.redis_study.use;import com.study.study_demo_of_spring_boot.redis_study.config.MyRedissonConfig;
import com.study.study_demo_of_spring_boot.redis_study.util.RedisUtil;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.redisson.RedissonRedLock;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.util.concurrent.TimeUnit;/*** <p>* 功能描述：* </p>** @author cui haida* @date 2024/04/14/7:28*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class UseTest {@Autowired@Qualifier("redLock1")private RedissonClient redLock1;@Autowired@Qualifier("redLock2")private RedissonClient redLock2;@Autowired@Qualifier("redLock3")private RedissonClient redLock3;@Testpublic void redLockTest() {RLock lock1 = redLock1.getLock("global_lock_key");RLock lock2 = redLock2.getLock("global_lock_key");RLock lock3 = redLock3.getLock("global_lock_key");// 三個構成red lockRedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);//定義獲取鎖標志位,默認是獲取失敗boolean isLockBoolean = false;try {// 等待獲取鎖的最長時間。如果在等待時間內無法獲取鎖，并且沒有其他鎖釋放，則返回 false。如果 waitTime < 0，則無限期等待，直到獲得鎖定。int waitTime = 1;// 就是redis key的過期時間，鎖的持有時間，可以使用 ttl  查看過期時間。int leaseTime = 20;// 如果在持有時間結束前鎖未被釋放，則鎖將自動過期，沒有進行key續期，并且其他線程可以獲得此鎖。如果 leaseTime = 0，則鎖將永久存在，直到被顯式釋放。isLockBoolean = redLock.tryLock(waitTime, leaseTime, TimeUnit.SECONDS);System.out.printf("線程:"+Thread.currentThread().getId()+",是否拿到鎖:" +isLockBoolean +"\n");if (isLockBoolean) {System.out.println("線程:"+Thread.currentThread().getId() + ",加鎖成功，進入業務操作");try {//業務邏輯，40s模擬，超過了key的過期時間TimeUnit.SECONDS.sleep(40);} catch (InterruptedException e) {e.printStackTrace();}}} catch (Exception e) {System.err.printf("線程:"+Thread.currentThread().getId()+"發生異常,加鎖失敗");e.printStackTrace();} finally {// 無論如何，最后都要解鎖redLock.unlock();}System.out.println(isLockBoolean);}
}

3：讀寫鎖

基于Redis的Redisson分布式可重入讀寫鎖RReadWriteLock

Java對象實現了java.util.concurrent.locks.ReadWriteLock接口。其中讀鎖和寫鎖都繼承了RLock接口。

分布式可重入讀寫鎖允許同時有多個讀鎖和一個寫鎖處于加鎖狀態。

3.1：讀鎖lock.readLock()

@GetMapping("/read")
public String readValue() {// 聲明一個可重入讀寫鎖RReadWriteLock lock = redissonClient.getReadWriteLock("rw-lock");String s = "";//加讀鎖RLock rLock = lock.readLock();rLock.lock();try {System.out.println("讀鎖加鎖成功"+Thread.currentThread().getId());                                           // 拿到values = redisTemplate.opsForValue().get("writeValue");Thread.sleep(30000);} catch (Exception e) {e.printStackTrace();} finally {// 解鎖rLock.unlock();System.out.println("讀鎖釋放"+Thread.currentThread().getId());}return  s;
}

3.2：寫鎖lock.writeLock()

@GetMapping("/write")
public String writeValue(){// 獲取一把鎖RReadWriteLock lock = redissonClient.getReadWriteLock("rw-lock");String s = "";// 加寫鎖RLock rLock = lock.writeLock();try {//1、改數據加寫鎖，讀數據加讀鎖rLock.lock();System.out.println("寫鎖加鎖成功..."+Thread.currentThread().getId());s = UUID.randomUUID().toString();Thread.sleep(30000);// 寫入redis中redisTemplate.opsForValue().set("writeValue",s);} catch (Exception e) {e.printStackTrace();} finally {rLock.unlock();System.out.println("寫鎖釋放"+Thread.currentThread().getId());}return  s;
}

• 先加寫鎖，后加讀鎖，此時并不會立刻給數據加讀鎖，而是需要等待寫鎖釋放后，才能加讀鎖
• 先加讀鎖，再加寫鎖：有讀鎖，寫鎖需要等待
• 先加讀鎖，再加讀鎖：并發讀鎖相當于無鎖模式，會同時加鎖成功

只要有寫鎖的存在，都必須等待，寫鎖是一個排他鎖，只能有一個寫鎖存在，讀鎖是一個共享鎖，可以有多個讀鎖同時存在

源碼在這里：https://blog.csdn.net/meser88/article/details/116591953

4：Semaphore和countDownLatch

4.1：Semaphore

基本使用

基于Redis的Redisson的分布式信號量（Semaphore）

Java對象RSemaphore采用了與java.util.concurrent.Semaphore相似的接口和用法

Semaphore是信號量，可以設置許可的個數，表示同時允許多個線程使用這個信號量（acquire()獲取許可）

• 如果沒有許可可用就線程阻塞，并且通過AQS進行排隊
• 可以使用release()釋放許可，當某一個線程釋放了某一個許可之后，將會從AQS中依次喚醒，直到沒有空閑許可。

@Test
public void semaphoreTest() throws InterruptedException {RSemaphore semaphore = redissonClient.getSemaphore("semaphore");// 同時最多允許3個線程獲取鎖semaphore.trySetPermits(3);for(int i = 0; i < 10; i++) {new Thread(() -> {try {System.out.println(new Date() + "：線程[" + Thread.currentThread().getName() + "]嘗試獲取Semaphore鎖");semaphore.acquire();System.out.println(new Date() + "：線程[" + Thread.currentThread().getName() + "]成功獲取到了Semaphore鎖，開始工作");Thread.sleep(3000);semaphore.release();System.out.println(new Date() + "：線程[" + Thread.currentThread().getName() + "]釋放Semaphore鎖");} catch (Exception e) {e.printStackTrace();}}).start();}Thread.sleep(5000);
}

源碼分析 - trySetPermits

@Override
public boolean trySetPermits(int permits) {return get(trySetPermitsAsync(permits));
}@Override
public RFuture<Boolean> trySetPermitsAsync(int permits) {RFuture<Boolean> future = commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"local value = redis.call('get', KEYS[1]); " +"if (value == false or value == 0) then "+ "redis.call('set', KEYS[1], ARGV[1]); "+ "redis.call('publish', KEYS[2], ARGV[1]); "+ "return 1;"+ "end;"+ "return 0;",Arrays.asList(getRawName(), getChannelName()), permits);// other....完成的時候打日志
}

1. get semaphore，獲取到一個當前的值
2. 第一次數據為0， 然后使用set semaphore 3，將這個信號量同時能夠允許獲取鎖的客戶端的數量設置為3
3. 然后發布一些消息，返回1

源碼分析 -> acquire

@Override
public void acquire(int permits) throws InterruptedException {// try - acquire ?if (tryAcquire(permits)) {return;}RFuture<RedissonLockEntry> future = subscribe();commandExecutor.syncSubscriptionInterrupted(future);try {while (true) {if (tryAcquire(permits)) {return;}future.getNow().getLatch().acquire();}} finally {unsubscribe(future);}// get(acquireAsync(permits));
}@Override
public boolean tryAcquire(int permits) {return get(tryAcquireAsync(permits));
}@Override
public RFuture<Boolean> tryAcquireAsync(int permits) {if (permits < 0) {throw new IllegalArgumentException("Permits amount can't be negative");}if (permits == 0) {return RedissonPromise.newSucceededFuture(true);}return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"local value = redis.call('get', KEYS[1]); " +"if (value ~= false and tonumber(value) >= tonumber(ARGV[1])) then " +"local val = redis.call('decrby', KEYS[1], ARGV[1]); " +"return 1; " +"end; " +"return 0;",Collections.<Object>singletonList(getRawName()), permits);
}

1. get semaphore，獲取到一個當前的值，比如說是3，3 > 1
2. decrby semaphore 1，將信號量允許獲取鎖的客戶端的數量遞減1，變成2
3. decrby semaphore 1
4. decrby semaphore 1
5. 執行3次加鎖后，semaphore值為0

此時如果再來進行加鎖則直接返回0，然后進入死循環去獲取鎖

源碼分析 -> release

@Override
public RFuture<Void> releaseAsync(int permits) {if (permits < 0) {throw new IllegalArgumentException("Permits amount can't be negative");}if (permits == 0) {return RedissonPromise.newSucceededFuture(null);}RFuture<Void> future = commandExecutor.evalWriteAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,"local value = redis.call('incrby', KEYS[1], ARGV[1]); " +"redis.call('publish', KEYS[2], value); ",Arrays.asList(getRawName(), getChannelName()), permits);if (log.isDebugEnabled()) {future.onComplete((o, e) -> {if (e == null) {log.debug("released, permits: {}, name: {}", permits, getName());}});}return future;
}

1. incrby semaphore 1，每次一個客戶端釋放掉這個鎖的話，就會將信號量的值累加1，信號量的值就不是0了

4.2：閉鎖CountDownLatch

基于Redisson的Redisson分布式閉鎖（CountDownLatch）

Java對象RCountDownLatch采用了與java.util.concurrent.CountDownLatch相似的接口和用法。

countDownLatch是計數器，可以設置一個數字，一個線程如果調用countDownLatch的await()將會發生阻塞

其他的線程可以調用countDown()對數字進行減一，數字成為0之后，阻塞的線程就會被喚醒。

底層原理就是，調用了await()的方法會利用AQS進行排隊。一旦數字成為0。AQS中的內容將會被依次喚醒。

@Test
public void countDownLatchTest() throws InterruptedException {RCountDownLatch latch = redissonClient.getCountDownLatch("anyCountDownLatch");latch.trySetCount(3);System.out.println(new Date() + "：線程[" + Thread.currentThread().getName() + "]設置了必須有3個線程執行countDown，進入等待中。。。");for(int i = 0; i < 3; i++) {new Thread(() -> {try {System.out.println(new Date() + "：線程[" + Thread.currentThread().getName() + "]在做一些操作，請耐心等待。。。。。。");Thread.sleep(3000);RCountDownLatch localLatch = redissonClient.getCountDownLatch("anyCountDownLatch");localLatch.countDown();System.out.println(new Date() + "：線程[" + Thread.currentThread().getName() + "]執行countDown操作");} catch (Exception e) {e.printStackTrace();}}).start();}// 一等多模型，主線程阻塞等子線程執行完畢，將countdown -> 0，主線程才能往下走latch.await();System.out.println(new Date() + "：線程[" + Thread.currentThread().getName() + "]收到通知，有3個線程都執行了countDown操作，可以繼續往下走");
}

先分析 trySetCount() 方法邏輯：

@Override
public RFuture<Boolean> trySetCountAsync(long count) {return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"if redis.call('exists', KEYS[1]) == 0 then "+ "redis.call('set', KEYS[1], ARGV[2]); "+ "redis.call('publish', KEYS[2], ARGV[1]); "+ "return 1 "+ "else "+ "return 0 "+ "end",Arrays.<Object>asList(getName(), getChannelName()), newCountMessage, count);
}

1. exists anyCountDownLatch，第一次肯定是不存在的
2. set redisson_countdownlatch__channel__anyCountDownLatch 3
3. 返回1

接著分析 latch.await()方法

@Override
public void await() throws InterruptedException {if (getCount() == 0) {return;}RFuture<RedissonCountDownLatchEntry> future = subscribe();try {commandExecutor.syncSubscriptionInterrupted(future);while (getCount() > 0) {// waiting for open statefuture.getNow().getLatch().await();}} finally {unsubscribe(future);}
}

這個方法其實就是陷入一個while true死循環，不斷的get anyCountDownLatch的值

如果這個值還是大于0那么就繼續死循環，否則的話呢，就退出這個死循環

最后分析 localLatch.countDown()方法

@Override
public RFuture<Void> countDownAsync() {return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"local v = redis.call('decr', KEYS[1]);" +"if v <= 0 then redis.call('del', KEYS[1]) end;" +"if v == 0 then redis.call('publish', KEYS[2], ARGV[1]) end;",Arrays.<Object>asList(getName(), getChannelName()), zeroCountMessage);
}

decr anyCountDownLatch，就是每次一個客戶端執行countDown操作，其實就是將這個cocuntDownLatch的值遞減1