Verwenden Sie Redis, um eine sichere und zuverlässige verteilte Sperre zu implementieren

In diesem Artikel erfahren Sie, wie Sie mit Redis eine sichere und zuverlässige verteilte Sperre implementieren und die Hauptelemente und häufigen Missverständnisse bei der Implementierung verteilter Sperren erläutern. Es hat einen gewissen Referenzwert. Freunde in Not können sich darauf beziehen. Ich hoffe, es wird für alle hilfreich sein.

In einem gleichzeitigen Szenario, in dem mehrere Prozesse oder Threads Ressourcen gemeinsam nutzen, muss sichergestellt werden, dass der Zugriff auf Ressourcen gegenseitig ausgeschlossen wird. In einem eigenständigen System können wir die API im Java-Parallelitätspaket, das synchronisierte Schlüsselwort usw. verwenden, um das Problem zu lösen. In einem verteilten System sind diese Methoden jedoch nicht mehr anwendbar und wir müssen verteilte Sperren selbst implementieren .

Zu den gängigen Lösungen zur Implementierung verteilter Sperren gehören: datenbankbasiert, Redis-basiert, Zookeeper-basiert usw. Im Rahmen des Redis-Themas wird in diesem Artikel über die Implementierung verteilter Sperren auf Basis von Redis gesprochen. [Verwandte Empfehlungen: Redis-Video-Tutorial]

Analyse und Implementierung

---

Problemanalyse

Verteilte Sperren und integrierte JVM-Sperren haben den gleichen Zweck: Anwendungen den erwarteten Zugriff oder Betrieb zu ermöglichen Ordnung Gemeinsam genutzte Ressourcen verhindern, dass mehrere Threads gleichzeitig auf derselben Ressource arbeiten, was zu einem chaotischen und unkontrollierbaren Systembetrieb führt. Wird häufig in Szenarien wie Produktbestandsabzügen und Couponabzügen verwendet.

Um die Sicherheit und Wirksamkeit der Sperre zu gewährleisten, müssen verteilte Sperren theoretisch mindestens die folgenden Bedingungen erfüllen:

• Gegenseitige Exklusivität: Nur ein Thread kann die Sperre gleichzeitig erhalten
• Kein Deadlock: Nachdem der Thread die Sperre erhalten hat, muss gewährleistet sein, dass er freigegeben wird. Selbst wenn die Anwendung nach Erhalt der Sperre ausfällt, kann sie innerhalb einer begrenzten Zeit freigegeben werden
In Bezug auf die Implementierung ist die Verteilungssperre grob in drei Schritte unterteilt:

a – Betriebsrechte für Ressourcen erhalten;

b – Vorgänge für Ressourcen ausführen;

• Unabhängig davon, ob es sich um die in Java integrierte Sperre oder um verteilte Sperren handelt, dreht sich alles um zwei Schritte a und c, unabhängig davon, welches verteilte Implementierungsschema verwendet wird. Redis ist aus folgenden Gründen von Natur aus für die Implementierung verteilter Sperren geeignet:
• Redis verwendet Single-Thread-Verarbeitung während der Befehlsverarbeitungsphase. Nur ein Thread kann denselben Schlüssel gleichzeitig verarbeiten, sodass keine Multi-Thread-Race-Bedingung vorliegt Problem. Der Befehl

SET key value NX PX milliseconds fügt einen Schlüssel mit Ablaufzeit hinzu, wenn der Schlüssel nicht vorhanden ist, und bietet so Unterstützung für Sicherheitsverriegelung.

Lua-Skripte und DEL-Befehle bieten zuverlässige Unterstützung für das sichere Entsperren.

Code-Implementierung
• SET key value NX PX milliseconds命令在不存在key的情况下添加具有过期时间的key，为安全加锁提供支持。
• Lua脚本和DEL命令为安全解锁提供可靠支撑。

代码实现

• Maven依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
  	<version>${your-spring-boot-version}</version>
</dependency>

• 配置文件

在application.properties增加以下内容，单机版Redis实例。

spring.redis.database=0
spring.redis.host=localhost
spring.redis.port=6379

• RedisConfig

@Configuration
public class RedisConfig {

    // 自己定义了一个 RedisTemplate
    @Bean
    @SuppressWarnings("all")
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory)
        throws UnknownHostException {
        // 我们为了自己开发方便，一般直接使用 <String, Object>
        RedisTemplate<String, Object> template = new RedisTemplate<String,
            Object>();
        template.setConnectionFactory(factory);
        // Json序列化配置
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);

        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // String 的序列化
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        // key采用String的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        // hash的key也采用String的序列化方式
        template.setHashKeySerializer(stringRedisSerializer);
        // value序列化方式采用jackson
        template.setValueSerializer(jackson2JsonRedisSerializer);
        // hash的value序列化方式采用jackson
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();
        return template;
    }
}

• RedisLock

@Service
public class RedisLock {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 加锁，最多等待maxWait毫秒
     *
     * @param lockKey   锁定key
     * @param lockValue 锁定value
     * @param timeout   锁定时长（毫秒）
     * @param maxWait   加锁等待时间（毫秒）
     * @return true-成功，false-失败
     */
    public boolean tryAcquire(String lockKey, String lockValue, int timeout, long maxWait) {
        long start = System.currentTimeMillis();

        while (true) {
            // 尝试加锁
            Boolean ret = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, timeout, TimeUnit.MILLISECONDS);
            if (!ObjectUtils.isEmpty(ret) && ret) {
                return true;
            }

            // 计算已经等待的时间
            long now = System.currentTimeMillis();
            if (now - start > maxWait) {
                return false;
            }

            try {
                Thread.sleep(200);
            } catch (Exception ex) {
                return false;
            }
        }
    }

    /**
     * 释放锁
     *
     * @param lockKey   锁定key
     * @param lockValue 锁定value
     * @return true-成功，false-失败
     */
    public boolean releaseLock(String lockKey, String lockValue) {
        // 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";

        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(script, Long.class);
        Long result = redisTemplate.opsForValue().getOperations().execute(redisScript, Collections.singletonList(lockKey), lockValue);
        return result != null && result > 0L;
    }
}

• 测试用例

@SpringBootTest
class RedisDistLockDemoApplicationTests {

    @Resource
    private RedisLock redisLock;

    @Test
    public void testLock() {
        redisLock.tryAcquire("abcd", "abcd", 5 * 60 * 1000, 5 * 1000);
        redisLock.releaseLock("abcd", "abcd");
    }
}

安全隐患

可能很多同学（也包括我）在日常工作中都是使用上面的实现方式，看似是稳妥的：

• 使用set命令NX、PX选项进行加锁，保证了加锁互斥，避免了死锁；
• 使用lua脚本解锁，防止解除其他线程的锁；
• 加锁、解锁命令都是原子操作；

其实以上实现的稳妥有个前提条件：单机版Redis、开启AOF持久化方式并设置appendfsync=always

Maven-Abhängigkeit

    @Test
    public void testRedLock() throws InterruptedException {

        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        final RedissonClient client = Redisson.create(config);

        // 获取锁实例
        final RLock lock = client.getLock("test-lock");

        // 加锁
        lock.lock(60 * 1000, TimeUnit.MILLISECONDS);
        try {
            // 假装做些什么事情
            Thread.sleep(50 * 1000);
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            //解锁
            lock.unlock();
        }
    }

Konfigurationsdatei

Fügen Sie den folgenden Inhalt in application.properties, eigenständige Redis-Instanz, hinzu.

rrreeeRedisConfig

---

rrreee

RedisLock

rrreee

Testfällerrreee

Sicherheitsrisiken

Vielleicht sind es viele Studenten (einschließlich mir). Mach es rein ihre tägliche Arbeit Die obige Implementierungsmethode wird in allen verwendet, was sicher zu sein scheint:

Verwenden Sie die Befehle set NX und PX Optionen zum Sperren, gegenseitiger Ausschluss von Sperren ist garantiert und ein Deadlock wird vermieden. 🎜🎜Verwenden Sie ein Lua-Skript, um das Entsperren anderer Threads zu verhindern Voraussetzung dafür, dass die obige Implementierung stabil ist: Standalone-Version von Redis, AOF-Persistenzmodus aktivieren und appendfsync=always festlegen. 🎜🎜Aber es kann zu Problemen im Sentry-Modus und im Cluster-Modus kommen, warum? 🎜🎜Sentinel-Modus und Cluster-Modus basieren auf der Master-Slave-Architektur. Die Datensynchronisierung zwischen Master und Slave erfolgt durch Befehlsweitergabe, und die Befehlsweitergabe erfolgt asynchron. 🎜🎜Es besteht also die Möglichkeit, dass die Daten des Masterknotens erfolgreich geschrieben wurden, der Masterknoten jedoch ausfällt, ohne den Slaveknoten zu benachrichtigen. 🎜🎜Wenn der Slave-Knoten durch Failover zum neuen Master-Knoten befördert wird, haben andere Threads die Möglichkeit, sich erfolgreich erneut zu sperren, was dazu führt, dass die gegenseitige Ausschlussbedingung der verteilten Sperre nicht erfüllt ist. 🎜🎜🎜Offizieller RedLock🎜🎜🎜🎜Wenn im Cluster-Modus alle Knoten im Cluster stabil laufen und kein Failover auftritt, ist die Sicherheit gewährleistet. Allerdings kann kein System 100 % Stabilität garantieren und verteilte Sperren auf Basis von Redis müssen Fehlertoleranz berücksichtigen. 🎜🎜🎜Da die Master-Slave-Synchronisation auf dem Prinzip der asynchronen Replikation basiert, können der Sentry-Modus und der Cluster-Modus diese Bedingung von Natur aus nicht erfüllen. Aus diesem Grund hat der Redis-Autor speziell eine Lösung vorgeschlagen: RedLock (Redis Distribute Lock). 🎜🎜🎜Designideen🎜🎜🎜Laut offiziellem Dokument werden die Designideen von RedLock vorgestellt. 🎜🎜Lassen Sie uns zunächst über die Umgebungsanforderungen sprechen. Es sind keine unabhängig voneinander bereitgestellten Redis-Instanzen erforderlich. Failover und andere Technologien sind nicht erforderlich. 🎜🎜Um die Sperre zu erhalten, befolgt der Kunde den folgenden Prozess: 🎜

• 获取当前时间（毫秒）作为开始时间start；
• 使用相同的key和随机value，按顺序向所有N个节点发起获取锁的请求。当向每个实例设置锁时，客户端会使用一个过期时间（小于锁的自动释放时间）。比如锁的自动释放时间是10秒，这个超时时间应该是5-50毫秒。这是为了防止客户端在一个已经宕机的实例浪费太多时间：如果Redis实例宕机，客户端尽快处理下一个实例。
• 客户端计算加锁消耗的时间cost（cost=start-now）。只有客户端在半数以上实例加锁成功，并且整个耗时小于整个有效时间（ttl），才能认为当前客户端加锁成功。
• 如果客户端加锁成功，那么整个锁的真正有效时间应该是：validTime=ttl-cost。
• 如果客户端加锁失败（可能是获取锁成功实例数未过半，也可能是耗时超过ttl），那么客户端应该向所有实例尝试解锁（即使刚刚客户端认为加锁失败）。

RedLock的设计思路延续了Redis内部多种场景的投票方案，通过多个实例分别加锁解决竞态问题，虽然加锁消耗了时间，但是消除了主从机制下的安全问题。

代码实现

官方推荐Java实现为Redisson，它具备可重入特性，按照RedLock进行实现，支持独立实例模式、集群模式、主从模式、哨兵模式等；API比较简单，上手容易。示例如下（直接通过测试用例）：

    @Test
    public void testRedLock() throws InterruptedException {

        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        final RedissonClient client = Redisson.create(config);

        // 获取锁实例
        final RLock lock = client.getLock("test-lock");

        // 加锁
        lock.lock(60 * 1000, TimeUnit.MILLISECONDS);
        try {
            // 假装做些什么事情
            Thread.sleep(50 * 1000);
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            //解锁
            lock.unlock();
        }
    }

Redisson封装的非常好，我们可以像使用Java内置的锁一样去使用，代码简洁的不能再少了。关于Redisson源码的分析，网上有很多文章大家可以找找看。

全文总结

---

分布式锁是我们研发过程中常用的的一种解决并发问题的方式，Redis是只是一种实现方式。

关键的是要弄清楚加锁、解锁背后的原理，以及实现分布式锁需要解决的核心问题，同时考虑我们所采用的中间件有什么特性可以支撑。了解这些后，实现起来就不是什么问题了。

学习了RedLock的思想，我们是不是也可以在自己的应用程序内实现了分布式锁呢？欢迎沟通！

更多编程相关知识，请访问：编程入门！！

Das obige ist der detaillierte Inhalt vonVerwenden Sie Redis, um eine sichere und zuverlässige verteilte Sperre zu implementieren. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!