Redis实现分布式锁详解

WBOY 原创: 2023-06-21 11:02:38 635浏览

随着移动互联网的快速发展和数据量的爆炸式增长，分布式系统变得越来越普及。分布式系统中，并发操作的问题就变得越来越凸显，当多个线程同时请求共享资源时，就需要对这些资源进行加锁，保证数据的一致性。分布式锁是一种实现分布式系统并发操作的有效方案之一，本文将详细介绍如何使用 Redis 实现分布式锁。

Redis 基础

Redis 是一个基于内存的键值对存储系统，在分布式系统中被广泛使用。Redis 作为一种高性能的 NoSQL 数据库，以其高效的读写性能和丰富的数据结构而受到广泛关注。Redis 可以基于多个机器实现分布式存储，同时支持如下数据结构：

字符串（string）
哈希（hash）
列表（list）
集合（set）
有序集合（sorted set）

Redis 的操作都是基于这些数据结构，为实现分布式锁需要用到 Redis 的一个特性：SETNX（SET if Not eXists），即当指定的键不存在时，才能设置键的值。如果键已经存在，则 SETNX 操作会返回失败。

实现分布式锁的思路

要实现分布式锁，首先需要明确目标：

在分布式环境中，多个线程同时请求同一个资源时，要保证只有一个线程可以获得锁。
如果某个线程已经获得锁，其他线程则需要等待锁的释放。

为了实现上述目标，可以采用以下思路：

使用 Redis 的 SETNX 命令创建一个新的键，作为锁的标识。
如果 SETNX 命令返回成功，表示当前线程获得了锁。
设置键的过期时间，避免死锁的情况。
当某个线程完成任务后，释放锁，即删除该键。

实现代码示例

首先，创建一个 Redis 连接：

import redis

conn = redis.Redis(host='localhost', port=6379, db=0)

接着，定义获取锁和释放锁的函数：

def acquire_lock(conn, lockname, acquire_timeout=10, lock_timeout=10):
    identifier = str(uuid.uuid4())
    lockname = "lock:" + lockname
    end = time.time() + acquire_timeout
    while time.time() < end:
        if conn.setnx(lockname, identifier):
            conn.expire(lockname, lock_timeout)
            return identifier
        elif not conn.ttl(lockname):
            conn.expire(lockname, lock_timeout)
        time.sleep(0.001)
    return False

def release_lock(conn, lockname, identifier):
    pipe = conn.pipeline(True)
    lockname = "lock:" + lockname
    while True:
        try:
            pipe.watch(lockname)
            if pipe.get(lockname) == identifier:
                pipe.multi()
                pipe.delete(lockname)
                pipe.execute()
                return True
            pipe.unwatch()
            break
        except redis.exceptions.WatchError:
            pass
    return False

其中，acquire_lock 函数用于获取锁，参数说明如下：

conn：Redis 连接。
lockname：锁的名称。
acquire_timeout：获取锁时的超时时间，默认为 10 秒。
lock_timeout：锁的过期时间，默认为 10 秒。

该函数首先生成一个随机的标识符，然后每隔 0.001 秒尝试获取锁，并设置过期时间。如果在指定的超时时间内没有获取到锁，则返回 False。

release_lock 函数用于释放锁，参数说明如下：

conn：Redis 连接。
lockname：锁的名称。
identifier：获取锁时返回的标识符。

该函数首先使用 WATCH 命令监视锁，如果锁的值与标识符相同，则使用 MULTI 命令删除该锁，并执行操作。否则，终止监视并返回 False。

最后，使用 acquire_lock 和 release_lock 函数即可实现分布式锁的功能。示例代码如下：

import time
import uuid

def do_task():
    print("Task started...")
    time.sleep(5)
    print("Task finished")

def main():
    lockname = "mylock"
    identifier = acquire_lock(conn, lockname)
    if not identifier:
        print("Failed to obtain lock")
        return
    try:
        do_task()
    finally:
        release_lock(conn, lockname, identifier)

if __name__ == '__main__':
    main()

该示例代码中，使用 acquire_lock 函数获取锁，在执行任务后调用 release_lock 函数释放锁。