Lua+Redis：分布式锁解决方案分享

数据运维 2023-09-28 醒在深海的猫手机阅读

介绍

当我们涉及到多进程或多节点的分布式系统时，传统的单机锁机制不再足够应对并发控制的需求。这是因为在分布式环境中，多个进程或节点同时访问共享资源，传统锁无法有效地协调这种复杂的并发情况，这就引入了分布式锁，本文将一步一步引导大家使用lua脚本和redis实现分布式锁。

理解分布式锁

1、什么是分布式锁？

分布式锁的是确保在多个进程或多个节点之间对共享资源的访问是有序、互斥和原子的，以避免竞态条件和数据不一致性问题。在多进程或多节点环境中，分布式锁广泛应用于协调共享资源的安全访问。

2、Redis作为分布式锁的选择

Redis（Remote Dictionary Server）是一种高性能的开源内存数据库，因其具有以下优势，使其成为实现分布式锁的理想选择：

高性能和低延迟：Redis以内存为基础，使得数据的读写操作非常快速，具有极低的延迟，适用于高吞吐量的应用场景。
持久性支持：尽管Redis是内存数据库，但它支持不同级别的数据持久性，可以将数据持久化到磁盘，确保数据不会因服务器重启而丢失。
数据结构丰富：Redis支持多种数据结构，如字符串、哈希、列表、集合和有序集合等，这使得它非常适合在分布式锁实现中灵活地操作数据。
原子性操作：Redis提供了各种原子性操作，包括原子的SETNX（SET if Not eXists）操作，这是实现分布式锁所需的基本操作之一。
发布-订阅功能：Redis支持发布-订阅模式，可用于实现分布式锁的通知机制，以便其他进程能够获知锁的状态变化。
Lua脚本支持：Redis支持运行Lua脚本，这意味着可以在Redis服务器上执行复杂的原子性操作，确保在多个命令之间不会发生竞态条件(重点)。

Redis的高性能、持久性、丰富的数据结构以及对Lua脚本的支持，使其成为实现分布式锁的理想选择。特别是Lua脚本的原子性执行，确保了获取和释放分布式锁的操作是不可分割的，从而有效地解决了竞态条件问题，确保了分布式锁的可靠性。

Lua脚本基础

1、Lua脚本简介

Lua是一种轻量级、高性能的脚本语言，广泛用于嵌入式系统和游戏开发，也被用于各种其他应用中：

轻量级：Lua被设计为一种轻量级的脚本语言，具有小巧的代码库和低内存消耗。这使得它适用于嵌入式系统和资源受限的环境。
高性能：Lua的解释器非常快速，执行效率高。这使得它在需要快速执行的应用中表现出色，如游戏引擎。
可嵌入性：Lua可以轻松嵌入到其他编程语言中，例如C/C++。这种特性使得它成为扩展应用程序功能的有力工具。
简单的语法：Lua采用简单、清晰的语法，易于学习和使用。它支持面向过程和函数式编程范式。
动态类型：Lua是一种动态类型语言，变量的类型在运行时确定。这增加了灵活性，但也需要更多的注意力来处理类型相关的错误。

Redis是一种开源的内存数据库，常用于缓存、队列和实时数据处理等场景。Redis引入了Lua脚本引擎，允许用户编写和执行Lua脚本来操作Redis数据。Lua脚本可以在Redis服务器上执行，确保多个Redis命令在单个事务中原子执行。这对于需要执行多个命令来维护数据一致性的应用非常有用。

Lua脚本在Redis中的应用使得Redis不仅仅是一个简单的键值存储，还可以执行复杂的操作和自定义业务逻辑，提高了Redis的灵活性和性能。这使得它成为处理高并发、实时数据的流行选择。

2、Redis 执行 Lua脚本

Lua执行格式：EVAL script numkeys key [key ...] arg [arg ...]

 eval "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 key1 key2 arg1 arg2
 
eval: 脚本执行命令
 
"return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}": 脚本内容
 
2: key数量 
 
key1 key2 arg1 arg2: key和value的值  角标从1开始
 
因为numkeys数量为2，故key1 key2 为key， arg1 arg2为value

例1:

 127.0.0.1:6379> eval "return redis.call('set',KEYS[1],'liulianJAVA')" 1 name
OK
127.0.0.1:6379> get name
"liulianJAVA"
127.0.0.1:6379>

例2:

 127.0.0.1:6379> eval "return redis.call('set',KEYS[1], ARGV[1])" 1 name LIULIANJAVA
OK
127.0.0.1:6379> get name
"LIULIANJAVA"
127.0.0.1:6379>

第三部分：Java与Redis集成

1、Jedis库简介

Jedis是一个流行的Java客户端库，用于与Redis数据库进行通信。它提供了一组用于连接、执行Redis命令和操作数据的API，使Java开发人员能够轻松地与Redis服务器进行交互。以下是强调Jedis的重要性的几个方面：

简单易用的API：Jedis提供了直观且易于理解的API，使Java开发人员能够轻松地与Redis进行通信。这使得在Java应用程序中使用Redis变得非常容易。
高性能：Jedis被设计为高性能的Redis客户端库。它使用了连接池和管道技术来提高性能，从而在高并发环境中表现出色。这对于需要快速访问Redis的应用程序非常重要。
广泛的社区支持：Jedis是一个广泛采用的库，拥有庞大的开发者社区和资源。这意味着您可以轻松地找到文档、教程和解决方案，以解决与Jedis相关的问题。
Redis功能的完全支持：Jedis支持Redis的所有功能，包括字符串、哈希、列表、集合、有序集合等数据结构。您可以使用Jedis执行所有Redis命令，而无需担心兼容性问题。
事务和管道支持：Jedis支持Redis的事务和管道功能。这允许您将多个命令组合成一个原子性操作，或者批量执行多个命令以提高性能。

Jedis是一个强大、易于使用且高性能的Java库，用于与Redis数据库进行通信。它为Java开发人员提供了一个便捷的工具，使他们能够利用Redis的强大功能来构建高性能、可扩展和可靠的应用程序。因此，对于需要使用Redis的Java应用程序来说，Jedis是一个不可或缺的工具。

2、连接Redis

首先，确保已经将Jedis库添加到Java项目的依赖中。可以使用Maven或Gradle等构建工具来添加Jedis依赖。

  
    redis.clients
    jedis
    3.6.1

 import redis.clients.jedis.Jedis;
 
public class JedisExample {
 
    public static void main(String[] args) {
        // 创建一个Jedis连接实例，连接到Redis服务器，默认端口为6379
        Jedis jedis = new Jedis("localhost", 6379);
        jedis.auth("liulianJAVA");
        // 执行一些Redis操作
        jedis.set("myKey", "Hello, Redis!");
        String value = jedis.get("myKey");
        System.out.println("Value for 'myKey': " + value);
        // 关闭连接
        jedis.close();
    }
}

在这个示例中，我们首先创建了一个Jedis实例，并指定了要连接的Redis服务器的主机名（localhost）和端口号（6379）。然后，我们使用set方法将一个键值对存储在Redis中，使用get方法检索该键的值，最后关闭了连接。

第四部分：使用Lua脚本实现分布式锁

1、获取分布式锁

 public static boolean acquireLock(Jedis jedis, String lockName, int lockTimeout) {
        String luaScript = "if redis.call('set', KEYS[1], ARGV[1], 'NX', 'EX', ARGV[2]) then return 1 else return 0 end";
        Object result = jedis.eval(luaScript, 1, lockName, LOCK_VALUE, String.valueOf(lockTimeout));
        return result != null && "1".equals(result.toString());
    }

 获取锁所需参数:
jedis: redis连接   
lockName: 锁名称 
lockTimeout: 锁超时时间

2、释放分布式锁

 public static void releaseLock(Jedis jedis, String lockName) {
    jedis.del(lockName);
}

 释放锁所需参数: 
jedis: redis连接   
lockName:  锁名称

第五部分：锁示例和实践

1、分布式锁示例

LockUtil.java：

 import redis.clients.jedis.Jedis;
 
public class LockUtil {
    private static final String LOCK_VALUE = "locked"; // 锁的值
 
    // 获取锁 所需参数:jedis:redis连接信息    lockName: 锁名称 lockTimeout: 锁超时时间
    public static boolean acquireLock(Jedis jedis, String lockName, int lockTimeout) {
        String luaScript = "if redis.call('set', KEYS[1], ARGV[1], 'NX', 'EX', ARGV[2]) then return 1 else return 0 end";
        Object result = jedis.eval(luaScript, 1, lockName, LOCK_VALUE, String.valueOf(lockTimeout));
        return result != null && "1".equals(result.toString());
    }
 
    public static boolean acquireLockWithRetry(Jedis jedis, String LOCK_KEY, int maxRetrySeconds) {
        long startTime = System.currentTimeMillis();
        while (true) {
            long currentTime = System.currentTimeMillis();
            if (currentTime - startTime >= maxRetrySeconds * 1000L) {
                // 超过预设的秒数，返回获取失败
                return false;
            }
 
            // 尝试获取锁
            boolean lockAcquired = acquireLock(jedis,LOCK_KEY, 60000);
            if (lockAcquired) {
                return true; // 成功获取锁
            }
 
            // 等待一段时间后重试
            try {
                Thread.sleep(100); // 100毫秒后重试
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false; // 线程被中断，返回获取失败
            }
        }
    }
 
 
    // 释放锁所需参数: jedis: redis连接   lockName: 锁名称
    public static void releaseLock(Jedis jedis, String lockName) {
        jedis.del(lockName);
    }
}

2、实践模拟

Test.java：

 import java.util.ArrayList;
 
public class Test {
 
    private static Integer number = 10000;
 
    public static void main(String[] args) {
 
        ArrayList threads = new ArrayList();
        for (int i = 0; i  Test.number = Test.number -1));
        }
 
        for (int i = 0; i  {
            try {
                thread.join();
            } catch (InterruptedException ignored) {
            }
        });
         System.out.println("number结果: " + number);
 
    }
}

number结果: 241

引入分布式锁后:

 import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
 
import java.util.ArrayList;
import java.util.List;
 
public class DistributedLockExample {
 
    private static Integer number = 1000;
 
    public static void main(String[] args) {
        List threads = new ArrayList();
        // 创建Jedis连接池
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(1000);
        JedisPool jedisPool = new JedisPool(poolConfig, "127.0.0.1", 6379, 200000, "liulian");
        String lockName = "liulian";
 
 
        for (int i = 0; i  {
                try (Jedis jedis = jedisPool.getResource()) {
                    // 获取锁
                    boolean lockAcquired = LockUtil.acquireLockWithRetry(jedis, lockName, 600);
                    if (lockAcquired) {
                        try {
                            // 在锁内执行关键操作
                            number = number - 1;
                        } finally {
                            // 释放锁
                            LockUtil.releaseLock(jedis, lockName);
                        }
                    } else {
                        System.out.println("Failed to acquire lock.");
                    }
                }
            }));
        }
 
        threads.forEach(Thread::start);
 
        threads.forEach(thread -> {
            try {
                thread.join();
            } catch (InterruptedException ignored) {
            }
        });
 
        System.out.println("number结果: " + number);
 
        // 关闭Jedis连接池
        jedisPool.close();
    }
 
    private static void performCriticalOperation() {
        // 在这里执行关键操作，例如访问共享资源或执行任务
        System.out.println("Critical operation performed.");
    }
}

number结果: 0

分布式锁在现代分布式系统中扮演了重要的角色，它们确保了并发操作的安全性和一致性。通过使用Redis和Lua脚本，我们已经了解了如何实现一个简单但有效的分布式锁。然而，分布式系统中的锁管理不仅仅局限于此。在实际应用中，我们可能需要处理更多的复杂情况，如锁的超时、死锁检测、锁的可重入性等。

此外，分布式锁的性能和可用性也是关键因素，需要仔细考虑和优化。在实际生产环境中使用分布式锁时，请确保全面测试和监控，以确保系统的稳定性和性能。

无论如何，分布式锁是构建可靠分布式系统的关键组成部分，对于确保数据完整性和一致性至关重要。希望本文能够帮助您更好地理解分布式锁的基本原理和实现方式，并为大家在构建分布式应用程序时提供有价值的指导。

	eval "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 key1 key2 arg1 arg2

	eval: 脚本执行命令

	"return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}": 脚本内容

	2: key数量

	key1 key2 arg1 arg2: key和value的值角标从1开始

	因为numkeys数量为2，故key1 key2 为key， arg1 arg2为value

	127.0.0.1:6379> eval "return redis.call('set',KEYS[1],'liulianJAVA')" 1 name
	OK
	127.0.0.1:6379> get name
	"liulianJAVA"
	127.0.0.1:6379>

	import redis.clients.jedis.Jedis;

	public class JedisExample {

	public static void main(String[] args) {
	// 创建一个Jedis连接实例，连接到Redis服务器，默认端口为6379
	Jedis jedis = new Jedis("localhost", 6379);
	jedis.auth("liulianJAVA");
	// 执行一些Redis操作
	jedis.set("myKey", "Hello, Redis!");
	String value = jedis.get("myKey");
	System.out.println("Value for 'myKey': " + value);
	// 关闭连接
	jedis.close();
	}
	}

	public static boolean acquireLock(Jedis jedis, String lockName, int lockTimeout) {
	String luaScript = "if redis.call('set', KEYS[1], ARGV[1], 'NX', 'EX', ARGV[2]) then return 1 else return 0 end";
	Object result = jedis.eval(luaScript, 1, lockName, LOCK_VALUE, String.valueOf(lockTimeout));
	return result != null && "1".equals(result.toString());
	}

	获取锁所需参数:
	jedis: redis连接
	lockName: 锁名称
	lockTimeout: 锁超时时间

	public static void releaseLock(Jedis jedis, String lockName) {
	jedis.del(lockName);
	}

	import redis.clients.jedis.Jedis;

	public class LockUtil {
	private static final String LOCK_VALUE = "locked"; // 锁的值

	// 获取锁所需参数:jedis:redis连接信息 lockName: 锁名称 lockTimeout: 锁超时时间
	public static boolean acquireLock(Jedis jedis, String lockName, int lockTimeout) {
	String luaScript = "if redis.call('set', KEYS[1], ARGV[1], 'NX', 'EX', ARGV[2]) then return 1 else return 0 end";
	Object result = jedis.eval(luaScript, 1, lockName, LOCK_VALUE, String.valueOf(lockTimeout));
	return result != null && "1".equals(result.toString());
	}

	public static boolean acquireLockWithRetry(Jedis jedis, String LOCK_KEY, int maxRetrySeconds) {
	long startTime = System.currentTimeMillis();
	while (true) {
	long currentTime = System.currentTimeMillis();
	if (currentTime - startTime >= maxRetrySeconds * 1000L) {
	// 超过预设的秒数，返回获取失败
	return false;
	}

	// 尝试获取锁
	boolean lockAcquired = acquireLock(jedis,LOCK_KEY, 60000);
	if (lockAcquired) {
	return true; // 成功获取锁
	}

	// 等待一段时间后重试
	try {
	Thread.sleep(100); // 100毫秒后重试
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	return false; // 线程被中断，返回获取失败
	}
	}
	}


	// 释放锁所需参数: jedis: redis连接 lockName: 锁名称
	public static void releaseLock(Jedis jedis, String lockName) {
	jedis.del(lockName);
	}
	}

	import java.util.ArrayList;

	public class Test {

	private static Integer number = 10000;

	public static void main(String[] args) {

	ArrayList threads = new ArrayList();
	for (int i = 0; i Test.number = Test.number -1));
	}

	for (int i = 0; i {
	try {
	thread.join();
	} catch (InterruptedException ignored) {
	}
	});
	System.out.println("number结果: " + number);

	}
	}

	import redis.clients.jedis.Jedis;
	import redis.clients.jedis.JedisPool;
	import redis.clients.jedis.JedisPoolConfig;

	import java.util.ArrayList;
	import java.util.List;

	public class DistributedLockExample {

	private static Integer number = 1000;

	public static void main(String[] args) {
	List threads = new ArrayList();
	// 创建Jedis连接池
	JedisPoolConfig poolConfig = new JedisPoolConfig();
	poolConfig.setMaxTotal(1000);
	JedisPool jedisPool = new JedisPool(poolConfig, "127.0.0.1", 6379, 200000, "liulian");
	String lockName = "liulian";


	for (int i = 0; i {
	try (Jedis jedis = jedisPool.getResource()) {
	// 获取锁
	boolean lockAcquired = LockUtil.acquireLockWithRetry(jedis, lockName, 600);
	if (lockAcquired) {
	try {
	// 在锁内执行关键操作
	number = number - 1;
	} finally {
	// 释放锁
	LockUtil.releaseLock(jedis, lockName);
	}
	} else {
	System.out.println("Failed to acquire lock.");
	}
	}
	}));
	}

	threads.forEach(Thread::start);

	threads.forEach(thread -> {
	try {
	thread.join();
	} catch (InterruptedException ignored) {
	}
	});

	System.out.println("number结果: " + number);

	// 关闭Jedis连接池
	jedisPool.close();
	}

	private static void performCriticalOperation() {
	// 在这里执行关键操作，例如访问共享资源或执行任务
	System.out.println("Critical operation performed.");
	}
	}