【redis】使用redisTemplate優雅地操作redis及使用redis實現分散式鎖
前言:
上篇已經介紹了redis及如何安裝和叢集redis,這篇介紹如何通過工具優雅地操作redis.
Long Long ago,程式猿們還在通過jedis來操作著redis,那時候的猿類,一個個累的沒日沒夜,重複的造著輪子,忙得沒時間陪家人,終於有一天猿類的春天來了,spring家族的redis template 解放了程式猿的雙手,於是猿類從使用Jedis石器時代的進入自動化時代...
redis template是對jedis的高度封裝,讓java對redis的操作更加簡單,甚至連小學生都可以駕馭...
在正式進入學習前,先給大家介紹一款Redis視覺化工具,個人感覺比
傳送門:http://www.treesoft.cn/dms.html
亦可百度搜treesoft,我不是託...
在正式學習之前,我們再來回顧一下Redis的支援儲存的五大資料型別:
分別為String（字串）、List（列表）、Set（集合）、Hash（雜湊）和 Zset（有序集合）
RedisTemplate中封裝了對5種資料結構的操作:
redisTemplate.opsForValue();//操作字串 redisTemplate.opsForHash();//操作hash redisTemplate.opsForList();//操作list redisTemplate.opsForSet();//操作set redisTemplate.opsForZSet();//操作有序set
StringRedisTemplate與RedisTemplate
-
兩者的關係是StringRedisTemplate繼承RedisTemplate。
-
兩者的資料是不共通的；也就是說StringRedisTemplate只能管理StringRedisTemplate裡面的資料，RedisTemplate只能管理RedisTemplate中的資料。
-
SDR預設採用的序列化策略有兩種，一種是String的序列化策略，一種是JDK的序列化策略。
StringRedisTemplate預設採用的是String的序列化策略，儲存的key和value都是採用此策略序列化儲存的。
RedisTemplate預設採用的是JDK的序列化策略，儲存的key和value都是採用此策略序列化儲存的。
以上兩種方式,根據實際業務需求靈活去選擇,操作字串型別用StringRedis Template,操作其它資料型別用Redis Template.
Redis Template的使用分為三步:引依賴,配置,使用...
第一步:引入依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>第二步:配置Redis Template(redisTemplate或StringRedisTemlate根據業務任選一種)
/**
* redis配置類
**/
@Configuration
@EnableCaching//開啟註解
public class RedisConfig {
//以下兩種redisTemplate自由根據場景選擇,優先推薦使用StringRedisTemplate
/**redisTemplate方式*/
@Bean
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
RedisTemplate<Object, Object> template = new RedisTemplate<>();
template.setConnectionFactory(connectionFactory);
//使用Jackson2JsonRedisSerializer來序列化和反序列化redis的value值（預設使用JDK的序列化方式）
Jackson2JsonRedisSerializer serializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper mapper = new ObjectMapper();
mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
serializer.setObjectMapper(mapper);
template.setValueSerializer(serializer);
//使用StringRedisSerializer來序列化和反序列化redis的key值
template.setKeySerializer(new StringRedisSerializer());
template.afterPropertiesSet();
return template;
}
/**StringRedisTemplate方式*/
// @Bean
// public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
// StringRedisTemplate stringRedisTemplate = new StringRedisTemplate();
// stringRedisTemplate.setConnectionFactory(factory);
// return stringRedisTemplate;
// }
}配置application.yml:
spring:
redis:
host: 192.168.1.1
password: 123456 # 沒密碼的話不用配
port: 6379
database: 10 #我這裡因為從視覺化工具裡發現10這個庫比較空,為了方便演示,所以配了10.第三步:使用
為了今後使用方便,其實你可以封裝一個RedisService,其功能有點類似JPA或者MyBatis這種,把需要對redis的存取操作封裝進去,當然這一步只是建議,封不封由你...
由於之前配置了redisTemplate及其子類，故需要使用@Resource註解進行呼叫.
@Resource
private RedisTemplate<String, Object> redisTemplate;//型別可根據實際情況走然後就可以根據redisTemplate進行各種資料操作了:
使用：redisTemplate.opsForValue().set("name","tom");
結果：redisTemplate.opsForValue().get("name") 輸出結果為tom
更多的我就不演示了,只要你對redis的5大資料型別的基本操作掌握即可輕鬆使用,,比較簡單,沒啥意思,如果感興趣可以參考這篇部落格,寫得十分詳細:
https://blog.csdn.net/ruby_one/article/details/79141940
下面我主要說一下前面提到的封裝RedisService,二話不說我先上程式碼為敬:
先寫介面RedisService:
/**Redis存取操作*/
public interface RedisService {
void set(String key,Object value);//無過期時間
void set(String key,Object value,Long timeOutSec);//帶過期時間,單位是秒,可以配.
Object get(String key);
}再寫實現類:
@Service
public class RedisServiceImpl implements RedisService {
@Resource
RedisTemplate<String, Object> redisTemplate;
@Override
public void set(String key, Object value) {
ValueOperations<String, Object> valueOperations = redisTemplate.opsForValue();
valueOperations.set(key, value);
}
@Override
public void set(String key, Object value, Long timeOutSec) {
ValueOperations<String, Object> valueOperations = redisTemplate.opsForValue();
valueOperations.set(key, value, timeOutSec, TimeUnit.SECONDS);
}
@Override
public Object get(String key) {
ValueOperations<String, Object> valueOperations = redisTemplate.opsForValue();
return valueOperations.get(key);
}
}呼叫:
隨便寫了兩個頁面,第一個頁面表單傳Key過來,第二個頁面對Key的value進行封裝並存入redis,再取出來作展現:
@RequestMapping("getValue")
public ModelAndView getValue(@RequestParam("key") String key, ModelAndView modelAndView) {
modelAndView.addObject("key", key);
User user = new User("老漢",18);
redisService.set(key,user,10L);
Object value = redisService.get(key);
modelAndView.addObject("value",value);
modelAndView.setViewName(PREFIX + "hello.html");
return modelAndView;
}
}效果:
然後我們進入TreeSoft來看一下redis中的資料是否有存進來:
可以看到,沒有問題,資料已經進來,10秒後再次重新整理頁面,資料已經過期,從redis資料庫中正常消失,完全符合預期.
前面提到了redisTemplate和StringRedisTemplate,下面我們看看他們除了我前面提到的那些差別,還有哪些地方不一樣:
重啟專案後,同樣的資料,看下效果:
結果未變,但redis中的資料變成了這樣...檢視不了,刪除不了,修改不了,因為亂碼了...看上去這種序列化方式似乎更加安全,但事實上,只是因為這款工具不支援顯示這樣的序列化方式編碼,換一個視覺化工具結果就不一樣了,所以不要被表面現象矇蔽了,要多文件及原始碼,兩者真正的差別是在操作資料型別上,StringRedisTemplate只適合操作String型別的,其他型別一律用RedisTemplate.
關於redis Template已是高度封裝了,對各種資料型別的操作都比較簡單,其他資料型別的操作我就不一一演示了,其實自從有了json,StringRedis Template 也可以用來儲存其他資料型別了,萬物皆字串,管你是什麼型別,都可以用Json字串來表示,所以大家重點掌握String型別的資料存取即可.
分散式鎖:
在單體應用架構中,遇到併發安全性問題時我們可以通過同步鎖Synchronized,同步程式碼塊,ReentrantLock等方式都可以解決,但在分散式系統中,JDK提供的這些併發鎖都失效了,我們需要一把"全域性的鎖",所有的分散式系統共享這把鎖,這把鎖同一時間內只能被一個系統擁有,擁有鎖的系統獲得一些相應的許可權,其它系統需要等待擁有鎖的系統釋放鎖,然後去競爭這把鎖,只有擁有這把鎖的系統才具有相應許可權.
分散式鎖目前比較常見的有3種實現方式,一種是基於Redis實現的,一種是基於zookeeper實現的,還有一種是基於資料庫層面的樂觀鎖和悲觀鎖.
本篇只介紹基於Redis的實現方式,其它兩種請翻閱本博,均有介紹和實現.
學之前先來了解一個將會用到的Redis命令
setNX(set if not exist):意思是如果不存在才會設定值,否則啥也不做,如果不存在,設定成功後返回值為1,失敗則返回0;
下面說一下實現原理:
1.所有系統在接收到請求後都去建立一把鎖,這把鎖的key均相同,但只有一個系統能最終建立成功,其他系統建立失敗.
2.建立鎖成功的系統繼續進行後續操作,比如下單,儲存資料至資料庫...未獲得鎖的系統等待,直到該系統操作完成後把鎖釋放,繼續開始競爭該鎖.為了確保分散式鎖可用，我們至少要確保鎖的實現同時滿足以下四個條件：
1.互斥性。在任意時刻，只有一個客戶端能持有鎖。
2.不會發生死鎖。即使有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證後續其他客戶端能加鎖。
3.具有容錯性。只要大部分的Redis節點正常執行，客戶端就可以加鎖和解鎖。
4.解鈴還須繫鈴人。加鎖和解鎖必須是同一個客戶端，客戶端自己不能把別人加的鎖給解了邏輯比較簡單,我直接上程式碼:
/**
*初始化Jedis連線池
*/
public class JedisPoolConfig {
private static JedisPool pool = null;
/**
* 靜態程式碼塊 構建redis連線池
*/
static {
if (pool == null) {
redis.clients.jedis.JedisPoolConfig config = new redis.clients.jedis.JedisPoolConfig();
//控制一個pool可分配多少個jedis例項，通過pool.getResource()來獲取；
//如果賦值為-1，則表示不限制；如果pool已經分配了maxActive個jedis例項，則此時pool的狀態為exhausted(耗盡)。
config.setMaxTotal(50);
//控制一個pool最多有多少個狀態為idle(空閒的)的jedis例項。
config.setMaxIdle(10);
//表示當borrow(引入)一個jedis例項時，最大的等待時間，如果超過等待時間，則直接丟擲JedisConnectionException；單位毫秒
//小於零:阻塞不確定的時間, 預設-1
config.setMaxWaitMillis(1000 * 100);
//在borrow(引入)一個jedis例項時，是否提前進行validate操作；如果為true，則得到的jedis例項均是可用的；
config.setTestOnBorrow(true);
//return 一個jedis例項給pool時，是否檢查連線可用性（ping()）
config.setTestOnReturn(true);
//connectionTimeout 連線超時（預設2000ms）
//soTimeout 響應超時（預設2000ms）
pool = new JedisPool(config, "192.168.1.1", 6379, 10000);
}
}
/**
* 方法描述 獲取Jedis例項
*
* @return
*/
public static Jedis getJedis() {
return pool.getResource();
}
/**
* 方法描述 釋放jedis連線資源
*
* @param jedis
*/
public static void returnResource(Jedis jedis) {
if (jedis != null) {
jedis.close();
}
}
}
public class DistributeLock {
private static final String LOCK_SUCCESS = "OK";
private static final String SET_IF_NOT_EXIST = "NX";
private static final String SET_WITH_EXPIRE_TIME = "PX";
private static final Long RELEASE_SUCCESS = 1L;
/**
* 嘗試獲取分散式鎖
* @param jedis Redis客戶端
* @param lockKey 鎖
* @param requestId 請求標識
* @param expireTime 超期時間
* @return 是否獲取成功
*/
public static boolean acquire(Jedis jedis, String lockKey, String requestId, int expireTime) {
String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
return false;
}
/**
* 釋放分散式鎖
* @param jedis Redis客戶端
* @param lockKey 鎖
* @param requestId 請求標識
* @return 是否釋放成功
*/
public static boolean release(Jedis jedis, String lockKey, String requestId) {
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(lockKey), Collections.singletonList(requestId));
if (RELEASE_SUCCESS.equals(result)) {
return true;
}
return false;
}
}在鎖的建立中,建立和設定過期時間必須保持原子性操作,否則萬一伺服器在建立鎖時宕機了,該節點變為永久節點,會造成死鎖.
在鎖的釋放中,判斷當前鎖是否有效和刪除該鎖也必須保持原子性操作,否則萬一伺服器在判斷鎖是否有效後發生GC或者其它卡頓,可能會造成誤刪,所以這裡用了Lua指令碼去執行,確保原子性.
另外上面有提到解鈴還須繫鈴人,故需要一個requestId來區分不同的請求.
原本想用redisTemplate來實現的,事實上我也確實用redisTemplate寫了一個,但因為自己不會寫lua指令碼,在鎖的釋放這裡不能做到原子性操作,所以借鑑了別人用Jedis方式的實現.
參考資料:https://www.cnblogs.com/linjiqin/p/8003838.html
https://blog.csdn.net/g6U8W7p06dCO99fQ3/article/details/81073892