Cara menggunakan Redis untuk melaksanakan kunci teragih dalam SpringBoot

1. Redis melaksanakan prinsip kunci teragih

Mengapa kunci teragih diperlukan

Sebelum bercakap tentang kunci teragih, perlu dijelaskan mengapa ia diperlukan 分布式锁.

Kebalikan daripada kunci yang diedarkan ialah kunci yang berdiri sendiri Apabila kami menulis program berbilang benang, kami mengelakkan masalah data yang disebabkan oleh mengendalikan pembolehubah yang dikongsi pada masa yang sama Kami biasanya menggunakan kunci untuk pengecualian bersama untuk memastikan ketepatan pembolehubah yang dikongsi, skop penggunaannya adalah dalam proses yang sama. Jika terdapat berbilang proses yang perlu mengendalikan sumber yang dikongsi pada masa yang sama, bagaimanakah ia boleh saling eksklusif? Aplikasi perniagaan hari ini biasanya merupakan seni bina perkhidmatan mikro, yang juga bermakna bahawa satu aplikasi akan menggunakan berbilang proses Jika berbilang proses perlu mengubah suai baris rekod yang sama dalam MySQL, untuk mengelakkan data kotor yang disebabkan oleh operasi yang tidak teratur, keperluan pengedaran. untuk diperkenalkan pada masa ini. Gaya dikunci.

Untuk melaksanakan kunci teragih, anda mesti menggunakan sistem luaran dan semua proses pergi ke sistem ini untuk memohon kunci. Sistem luaran ini mestilah saling eksklusif, iaitu, jika dua permintaan tiba pada masa yang sama, sistem hanya akan berjaya mengunci satu proses, dan proses yang lain akan gagal. Sistem luaran ini boleh menjadi pangkalan data, Redis atau Zookeeper, tetapi untuk mengejar prestasi, kami biasanya memilih untuk menggunakan Redis atau Zookeeper.

Redis boleh digunakan sebagai sistem storan kongsi dan berbilang pelanggan boleh berkongsi akses, jadi ia boleh digunakan untuk menyimpan kunci yang diedarkan. Selain itu, Redis mempunyai prestasi baca dan tulis yang tinggi dan boleh mengendalikan senario operasi kunci konkurensi tinggi. Fokus artikel ini adalah untuk memperkenalkan cara menggunakan Redis untuk melaksanakan kunci teragih, dan membincangkan masalah yang mungkin dihadapi semasa proses pelaksanaan.

Cara melaksanakan kunci teragih

Sebagai sistem storan kongsi dalam pelaksanaan kunci teragih, Redis boleh menggunakan pasangan nilai kunci untuk menyimpan pembolehubah kunci dan menerima serta memproses Mereka Permintaan operasi untuk mengunci dan melepaskan kunci yang dihantar oleh pelanggan yang berbeza. Jadi, bagaimanakah kunci dan nilai pasangan nilai kunci ditentukan? Kita perlu memberi pembolehubah kunci nama pembolehubah dan menggunakan nama pembolehubah ini sebagai kunci pasangan nilai kunci, dan nilai pembolehubah kunci ialah nilai pasangan nilai kunci Dengan cara ini, Redis boleh menyimpan pembolehubah kunci, dan klien boleh Operasi kunci boleh dilaksanakan melalui operasi arahan Redis.

Untuk melaksanakan kunci teragih, Redis mesti mempunyai keupayaan pengecualian bersama. Anda boleh menggunakan arahan SETNX, yang bermaksud SET JIKA TIDAK WUJUD, iaitu, jika kunci tidak wujud, nilainya akan ditetapkan, jika tidak, tiada apa yang akan dilakukan. Kunci yang diedarkan dilaksanakan dengan mempunyai dua proses klien melaksanakan perintah secara saling eksklusif.

Berikut menunjukkan proses operasi Redis menggunakan pasangan kunci/nilai untuk menyimpan pembolehubah kunci dan dua pelanggan meminta kunci pada masa yang sama.

Selepas operasi mengunci selesai, pelanggan yang telah berjaya mengunci boleh mengendalikan sumber yang dikongsi, contohnya, mengubah suai baris data tertentu dalam MySQL. Selepas operasi selesai, kunci mesti dilepaskan tepat pada masanya untuk memberi peluang kepada mereka yang lewat untuk mengendalikan sumber yang dikongsi. Bagaimana untuk melepaskan kunci? Hanya gunakan arahan DEL untuk memadam kunci ini. Logiknya sangat mudah Proses keseluruhan yang ditulis dalam kod pseudo adalah seperti berikut.

// 加锁
SETNX lock_key 1
// 业务逻辑
DO THINGS
// 释放锁
DEL lock_key

Namun, terdapat masalah besar dalam pelaksanaan di atas Apabila klien 1 mendapat kunci, jika senario berikut berlaku, kebuntuan akan berlaku.

Program ini mengendalikan pengecualian logik perniagaan dan gagal untuk melepaskan kunci dalam masa yang ditetapkan dan tidak mempunyai peluang untuk melepaskan kunci itu kunci untuk menduduki kunci selama-lamanya, dan pelanggan lain tidak akan dapat mendapatkannya.

Untuk menyelesaikan masalah kebuntuan di atas, penyelesaian yang paling mudah untuk difikirkan ialah menetapkan kunci semasa memohon kunci dan melaksanakannya dalam Redis Masa tamat, dengan mengandaikan bahawa masa untuk mengendalikan sumber yang dikongsi tidak akan melebihi 10 saat, maka apabila mengunci, hanya tetapkan masa tamat 10 saat untuk kunci ini.

Tetapi masih terdapat masalah dengan operasi di atas,

加锁、设置过期时间是2条命令，有可能只执行了第一条，第二条却执行失败

2. SETNX dilaksanakan Berjaya, Redis ranap secara tidak normal, EXPIRE tidak berpeluang untuk melaksanakan

3. SETNX telah berjaya dilaksanakan, pelanggan ranap secara tidak normal, EXPIRE tidak mempunyai peluang untuk melaksanakan

Pendek kata, ini ialah

两条命令如果不能保证是原子操作，就有潜在的风险导致过期时间设置失败，依旧有可能发生死锁问题

Pada ketika ini, masalah kebuntuan sudah selesai, tetapi masih ada masalah lain. Bayangkan senario berikut:

Klien 1 berjaya dikunci dan mula mengendalikan sumber kongsi

2. 客户端1操作共享资源耗时太久，超过了锁的过期时间，锁失效（锁被自动释放）

3. 客户端2加锁成功，开始操作共享资源

4. 客户端1操作共享资源完成，在finally块中手动释放锁，但此时它释放的是客户端2的锁。

这里存在两个严重的问题：

• 锁过期

• 释放了别人的锁

第1个问题是评估操作共享资源的时间不准确导致的，如果只是一味增大过期时间，只能缓解问题降低出现问题的概率，依旧无法彻底解决问题。原因在于客户端在拿到锁之后，在操作共享资源时，遇到的场景是很复杂的，既然是预估的时间，也只能是大致的计算，不可能覆盖所有导致耗时变长的场景。

第二个问题在于解锁操作是不够严谨的，因为它是一种不加区分地释放锁的操作，没有对锁的所有权进行检查。如何解决呢？

锁被别人给释放了

解决办法是，客户端在加锁时，设置一个只有自己知道的唯一标识进去，例如可以是自己的线程ID，如果是redis实现，就是SET key unique_value EX 10 NX。之后在释放锁时，要先判断这把锁是否归自己持有，只有是自己的才能释放它。

//释放锁 比较unique_value是否相等，避免误释放
if redis.get("key") == unique_value then
    return redis.del("key")

这里释放锁使用的是GET + DEL两条命令，这时又会遇到原子性问题了。

1. 客户端1执行GET，判断锁是自己的

2. 客户端2执行了SET命令，强制获取到锁（虽然发生概念很低，但要严谨考虑锁的安全性）

3. 客户端1执行DEL，却释放了客户端2的锁

由此可见，以上GET + DEL两个命令还是必须原子的执行才行。怎样原子执行两条命令呢？答案是Lua脚本，可以把以上逻辑写成Lua脚本，让Redis执行。因为Redis处理每个请求是单线程执行的，在执行一个Lua脚本时其它请求必须等待，直到这个Lua脚本处理完成，这样一来GET+DEL之间就不会有其他命令执行了。

以下是使用Lua脚本（unlock.script）实现的释放锁操作的伪代码，其中，KEYS[1]表示lock_key，ARGV[1]是当前客户端的唯一标识，这两个值都是我们在执行 Lua脚本时作为参数传入的。

//Lua脚本语言，释放锁 比较unique_value是否相等，避免误释放
if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

最后我们执行以下命令，即可

redis-cli  --eval  unlock.script lock_key , unique_value

这样一路优先下来，整个加锁、解锁流程就更严谨了，先小结一下，基于Redis实现的分布式锁，一个严谨的流程如下：

1. 加锁时要设置过期时间SET lock_key unique_value EX expire_time NX

2. 操作共享资源

3. 释放锁：Lua脚本，先GET判断锁是否归属自己，再DEL释放锁

有了这个严谨的锁模型，我们还需要重新思考之前的那个问题，锁的过期时间不好评估怎么办。

如何确定锁的过期时间

前面提到过，过期时间如果评估得不好，这个锁就会有提前过期的风险，一种妥协的解决方案是，尽量冗余过期时间，降低锁提前过期的概率，但这个方案并不能完美解决问题。是否可以设置这样的方案，加锁时，先设置一个预估的过期时间，然后开启一个守护线程，定时去检测这个锁的失效时间，如果锁快要过期了，操作共享资源还未完成，那么就自动对锁进行续期，重新设置过期时间。

Redisson是一个已封装好这些工作的库，可以说是一种非常优秀的解决方案。Redisson是一个Java语言实现的Redis SDK客户端，在使用分布式锁时，它就采用了自动续期的方案来避免锁过期，这个守护线程我们一般叫它看门狗线程。这个SDK提供的API非常友好，它可以像操作本地锁一样操作分布式锁。客户端一旦加锁成功，就会启动一个watch dog看门狗线程，它是一个后台线程，会每隔一段时间（这段时间的长度与设置的锁的过期时间有关）检查一下，如果检查时客户端还持有锁key（也就是说还在操作共享资源），那么就会延长锁key的生存时间。

那如果客户端在加锁成功后就宕机了呢？宕机了那么看门狗任务就不存在了，也就无法为锁续期了，锁到期自动失效。

Redis的部署方式对锁的影响

上面讨论的情况，都是锁在单个Redis 实例中可能产生的问题，并没有涉及到Redis的部署架构细节。

Redis发展到现在，几种常见的部署架构有：

• Mod tunggal;

• Mod tuan-hamba;

Mod kelompok;
Apabila kami menggunakan Redis,
• . Jadi apabila suis tuan-hamba berlaku, adakah kunci yang diedarkan masih selamat?

Bayangkan senario ini: 一般会采用主从集群+哨兵的模式部署，哨兵的作用就是监测redis节点的运行状态。普通的主从模式，当master崩溃时，需要手动切换让slave成为master，使用主从+哨兵结合的好处在于，当master异常宕机时，哨兵可以实现故障自动切换，把slave提升为新的master，继续提供服务，以此保证可用性

Klien 1 melaksanakan arahan SET pada induk dan kunci berjaya

Pada masa ini, tuan turun secara tidak normal, dan arahan SET belum lagi disegerakkan kepada hamba (replikasi tuan-hamba adalah tak segerak)
The sentinel mempromosikan hamba kepada tuan baharu, tetapi kunci telah hilang pada tuan baharu, menyebabkan pelanggan 2 berjaya mengunci , kunci yang diedarkan mungkin masih terjejas. Walaupun Redis memastikan ketersediaan tinggi melalui sentinel, jika nod induk menukar hamba induk atas sebab tertentu, kunci akan hilang.

2. Mod Kluster + Redlock melaksanakan kunci teragih yang sangat boleh dipercayai

3. Untuk mengelakkan masalah kegagalan kunci yang disebabkan oleh kegagalan contoh Redis, pemaju Redis Antirez mencadangkan algoritma kunci teragih Redlock . Idea asas algoritma Redlock,

   . Dengan cara ini, walaupun satu kejadian Redis gagal, kerana pembolehubah kunci juga disimpan pada kejadian lain, pelanggan masih boleh melakukan operasi kunci secara normal dan pembolehubah kunci tidak akan hilang.

Langkah pertama adalah untuk klien mendapatkan masa semasa.

Langkah kedua ialah untuk pelanggan melakukan operasi penguncian pada kejadian N Redis mengikut turutan. 是让客户端和多个独立的Redis实例依次请求加锁，如果客户端能够和半数以上的实例成功地完成加锁操作，那么我们就认为，客户端成功地获得分布式锁了，否则加锁失败

.

当然，如果客户端在和所有实例执行完加锁操作后，没能同时满足这两个条件，那么，客户端就要向所有Redis节点发起释放锁的操作。为什么释放锁，要操作所有的节点呢，不能只操作那些加锁成功的节点吗？因为在某一个Redis节点加锁时，可能因为网络原因导致加锁失败，例如一个客户端在一个Redis实例上加锁成功，但在读取响应结果时由于网络问题导致读取失败，那这把锁其实已经在Redis上加锁成功了。所以释放锁时，不管之前有没有加锁成功，需要释放所有节点上的锁以保证清理节点上的残留的锁。

在Redlock算法中，释放锁的操作和在单实例上释放锁的操作一样，只要执行释放锁的 Lua脚本就可以了。如果N个Redis实例中超过一半的实例正常工作，就能确保分布式锁正常运作。为了提高分布式锁的可靠性，您可以在实际业务应用中使用Redlock算法。

二、代码实现Redis分布式锁

1.SpringBoot整合redis用到最多的当然属于我们的老朋友RedisTemplate，pom依赖如下：

<!-- springboot整合redis -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

2.Redis配置类:

package com.example.redisdemo.config;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * @description: Redis配置类
 * @author Keson
 * @date 21:20 2022/11/14
 * @Param
 * @return
 * @version 1.0
 */
@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory) {
        // 设置序列化
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // 配置redisTemplate
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
        redisTemplate.setConnectionFactory(lettuceConnectionFactory);
        RedisSerializer<?> stringSerializer = new StringRedisSerializer();
        redisTemplate.setKeySerializer(stringSerializer);// key序列化
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);// value序列化
        redisTemplate.setHashKeySerializer(stringSerializer);// Hash key序列化
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);// Hash value序列化
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
}

3.Service层面

package com.example.redisdemo.service;

import com.example.redisdemo.entity.CustomerBalance;
import java.util.concurrent.Callable;

/**
 * @author Keson
 * @version 1.0
 * @description: TODO
 * @date 2022/11/14 15:12
 */
public interface RedisService {

    <T> T callWithLock(CustomerBalance customerBalance, Callable<T> callable) throws Exception;
}

package com.example.redisdemo.service.impl;

import com.example.redisdemo.entity.CustomerBalance;
import com.example.redisdemo.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * @author Keson
 * @version 1.0
 * @description: TODO Redis实现分布式锁
 * @date 2022/11/14 15:13
 */
@Service
@Slf4j
public class RedisServiceImpl implements RedisService {

    //设置默认过期时间
    private final static int DEFAULT_LOCK_EXPIRY_TIME = 20;
    //自定义lock key前缀
    private final static String LOCK_PREFIX = "LOCK:CUSTOMER_BALANCE";

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public <T> T callWithLock(CustomerBalance customerBalance, Callable<T> callable) throws Exception{
        //自定义lock key
        String lockKey = getLockKey(customerBalance.getCustomerNumber(), customerBalance.getSubAccountNumber(), customerBalance.getCurrencyCode());
        //将UUID当做value，确保唯一性
        String lockReference = UUID.randomUUID().toString();

        try {
            if (!lock(lockKey, lockReference, DEFAULT_LOCK_EXPIRY_TIME, TimeUnit.SECONDS)) {
                throw new Exception("lock加锁失败");
            }
            return callable.call();
        } finally {
            unlock(lockKey, lockReference);
        }
    }

    //定义lock key
    String getLockKey(String customerNumber, String subAccountNumber, String currencyCode) {
        return String.format("%s:%s:%s:%s", LOCK_PREFIX, customerNumber, subAccountNumber, currencyCode);
    }

    //redis加锁
    private boolean lock(String key, String value, long timeout, TimeUnit timeUnit) {
        Boolean locked;
        try {
            //SET_IF_ABSENT --> NX: Only set the key if it does not already exist.
            //SET_IF_PRESENT --> XX: Only set the key if it already exist.
            locked = (Boolean) redisTemplate.execute((RedisCallback<Boolean>) connection ->
                    connection.set(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8),
                            Expiration.from(timeout, timeUnit), RedisStringCommands.SetOption.SET_IF_ABSENT));
        } catch (Exception e) {
            log.error("Lock failed for redis key: {}, value: {}", key, value);
            locked = false;
        }
        return locked != null && locked;
    }

    //redis解锁
    private boolean unlock(String key, String value) {
        try {
            //使用lua脚本保证删除的原子性，确保解锁
            String script = "if redis.call(&#39;get&#39;, KEYS[1]) == ARGV[1] " +
                            "then return redis.call(&#39;del&#39;, KEYS[1]) " +
                            "else return 0 end";
            Boolean unlockState = (Boolean) redisTemplate.execute((RedisCallback<Boolean>) connection ->
                    connection.eval(script.getBytes(), ReturnType.BOOLEAN, 1,
                            key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8)));
            return unlockState == null || !unlockState;
        } catch (Exception e) {
            log.error("unLock failed for redis key: {}, value: {}", key, value);
            return false;
        }
    }
}

4.业务调用实现分布式锁示例：

    @Override
    public int updateById(CustomerBalance customerBalance) throws Exception {
        return redisService.callWithLock(customerBalance, ()-> customerBalanceMapper.updateById(customerBalance));
    }

Atas ialah kandungan terperinci Cara menggunakan Redis untuk melaksanakan kunci teragih dalam SpringBoot. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!