IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python Redis分布式锁案例如何实现锁机制

wen python案例 230

本文目录导读：

Python Redis分布式锁案例如何实现锁机制

  1. 基础实现：使用SETNX + EXPIRE
  2. 改进版本：带重试机制的分布式锁
  3. 使用RedLock算法（高可用方案）
  4. 实际业务应用示例
  5. 使用现成的库：redlock-py

我来详细讲解Python中使用Redis实现分布式锁的几种实现方式。

基础实现：使用SETNX + EXPIRE

import redis
import time
import uuid
class RedisLock:
    def __init__(self, redis_client, lock_key, timeout=10):
        self.redis = redis_client
        self.lock_key = lock_key
        self.timeout = timeout
        self.lock_value = str(uuid.uuid4())  # 唯一标识
        self.is_locked = False
    def acquire(self):
        """获取锁"""
        # 使用SET NX 和 EX 原子操作
        result = self.redis.set(
            self.lock_key,
            self.lock_value,
            nx=True,  # 只有key不存在时才设置
            ex=self.timeout  # 过期时间
        )
        if result:
            self.is_locked = True
            return True
        return False
    def release(self):
        """释放锁"""
        if self.is_locked:
            # 使用Lua脚本确保原子性
            lua_script = """
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end
            """
            self.redis.eval(lua_script, 1, self.lock_key, self.lock_value)
            self.is_locked = False
    def __enter__(self):
        self.acquire()
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()
# 使用示例
def use_basic_lock():
    r = redis.Redis(host='localhost', port=6379, db=0)
    lock = RedisLock(r, "my_lock")
    if lock.acquire():
        try:
            # 执行需要加锁的业务逻辑
            print("获取锁成功，执行关键操作")
            time.sleep(2)
        finally:
            lock.release()

改进版本：带重试机制的分布式锁

import redis
import time
import uuid
import random
class RedisDistributedLock:
    def __init__(self, redis_client, lock_key, timeout=10, retry_times=3, retry_delay=0.1):
        self.redis = redis_client
        self.lock_key = f"lock:{lock_key}"
        self.timeout = timeout
        self.retry_times = retry_times
        self.retry_delay = retry_delay
        self.lock_value = str(uuid.uuid4())
        self.is_locked = False
    def acquire(self, blocking=True):
        """
        获取锁
        :param blocking: 是否阻塞等待
        :return: bool
        """
        for attempt in range(self.retry_times if blocking else 1):
            result = self.redis.set(
                self.lock_key,
                self.lock_value,
                nx=True,
                ex=self.timeout
            )
            if result:
                self.is_locked = True
                return True
            if not blocking:
                return False
            # 退避策略：随机延迟
            delay = self.retry_delay * (2 ** attempt) + random.uniform(0, 0.1)
            time.sleep(delay)
        return False
    def release(self):
        """释放锁"""
        if self.is_locked:
            # Lua脚本保证原子性
            lua_script = """
            if redis.call('get', KEYS[1]) == ARGV[1] then
                redis.call('del', KEYS[1])
                return 1
            else
                return 0
            end
            """
            self.redis.eval(lua_script, 1, self.lock_key, self.lock_value)
            self.is_locked = False
    def extend_timeout(self, extra_time=10):
        """延长锁的超时时间"""
        if self.is_locked:
            lua_script = """
            if redis.call('get', KEYS[1]) == ARGV[1] then
                redis.call('expire', KEYS[1], ARGV[2])
                return 1
            else
                return 0
            end
            """
            return self.redis.eval(lua_script, 1, self.lock_key, self.lock_value, extra_time)
        return False
# 使用示例
def use_retry_lock():
    r = redis.Redis(host='localhost', port=6379, db=0)
    lock = RedisDistributedLock(r, "order_lock", timeout=30)
    if lock.acquire():
        try:
            print("处理订单...")
            # 如果业务处理时间较长，可以延长锁
            lock.extend_timeout(10)
            time.sleep(20)
        finally:
            lock.release()

使用RedLock算法（高可用方案）

import redis
import time
import uuid
class RedLock:
    """RedLock算法实现，需要多个Redis实例"""
    def __init__(self, redis_connections, lock_key, ttl=10000):
        """
        :param redis_connections: Redis连接列表
        :param lock_key: 锁的key
        :param ttl: 锁的存活时间(毫秒)
        """
        self.redis_connections = redis_connections
        self.lock_key = f"redlock:{lock_key}"
        self.ttl = ttl
        self.lock_value = str(uuid.uuid4())
        self.is_locked = False
    def acquire(self, retry_count=3, retry_delay=200):
        """
        获取锁
        :param retry_count: 重试次数
        :param retry_delay: 重试延迟(毫秒)
        :return: bool
        """
        for _ in range(retry_count):
            n = len(self.redis_connections)
            start_time = int(time.time() * 1000)
            locked_count = 0
            # 尝试在所有Redis实例上获取锁
            for redis_conn in self.redis_connections:
                try:
                    if redis_conn.set(
                        self.lock_key,
                        self.lock_value,
                        nx=True,
                        px=self.ttl  # 毫秒级过期
                    ):
                        locked_count += 1
                except redis.exceptions.RedisError:
                    pass
            # 计算获取锁的时间
            elapsed_time = int(time.time() * 1000) - start_time
            # 检查是否获取了大多数锁且没有超时
            if locked_count >= n // 2 + 1 and elapsed_time < self.ttl:
                self.is_locked = True
                return True
            # 释放已获取的锁
            for redis_conn in self.redis_connections:
                try:
                    lua_script = """
                    if redis.call('get', KEYS[1]) == ARGV[1] then
                        redis.call('del', KEYS[1])
                    end
                    """
                    redis_conn.eval(lua_script, 1, self.lock_key, self.lock_value)
                except redis.exceptions.RedisError:
                    pass
            time.sleep(retry_delay / 1000)
        return False
    def release(self):
        """释放锁"""
        if self.is_locked:
            for redis_conn in self.redis_connections:
                try:
                    lua_script = """
                    if redis.call('get', KEYS[1]) == ARGV[1] then
                        redis.call('del', KEYS[1])
                    end
                    """
                    redis_conn.eval(lua_script, 1, self.lock_key, self.lock_value)
                except redis.exceptions.RedisError:
                    pass
            self.is_locked = False
# 使用示例
def use_redlock():
    # 配置多个Redis实例
    redis_connections = [
        redis.Redis(host='localhost', port=6379, db=0),
        redis.Redis(host='localhost', port=6380, db=0),
        redis.Redis(host='localhost', port=6381, db=0)
    ]
    lock = RedLock(redis_connections, "resource_lock", ttl=10000)
    if lock.acquire():
        try:
            print("成功获取分布式锁")
            time.sleep(5)
        finally:
            lock.release()
            print("释放分布式锁")

实际业务应用示例

import redis
import json
import time
from contextlib import contextmanager
class DistributedLockService:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis = redis.Redis(
            host=redis_host,
            port=redis_port,
            decode_responses=True
        )
    @contextmanager
    def lock(self, lock_key, timeout=10, retry_times=3):
        """上下文管理器方式的锁"""
        lock = RedisDistributedLock(
            self.redis, 
            lock_key, 
            timeout=timeout,
            retry_times=retry_times
        )
        if lock.acquire():
            try:
                yield True
            finally:
                lock.release()
        else:
            yield False
    def process_order(self, order_id, user_id, amount):
        """处理订单的分布式锁示例"""
        lock_key = f"order:{order_id}"
        with self.lock(lock_key, timeout=30) as acquired:
            if not acquired:
                raise Exception(f"订单 {order_id} 正在处理中")
            try:
                # 模拟业务处理
                print(f"开始处理订单 {order_id}")
                # 查询订单状态
                order_status = self.redis.get(f"order_status:{order_id}")
                if order_status and order_status == "processing":
                    raise Exception("订单已经被处理")
                # 标记订单正在处理
                self.redis.set(f"order_status:{order_id}", "processing")
                # 执行实际业务逻辑
                time.sleep(2)  # 模拟耗时操作
                # 更新订单状态
                self.redis.set(f"order_status:{order_id}", "completed")
                self.redis.hset(f"order:{order_id}", "user", user_id)
                self.redis.hset(f"order:{order_id}", "amount", amount)
                print(f"订单 {order_id} 处理完成")
            except Exception as e:
                print(f"处理订单失败: {e}")
                self.redis.set(f"order_status:{order_id}", "failed")
                raise
    def inventory_deduction(self, product_id, quantity):
        """库存扣减的分布式锁示例"""
        lock_key = f"inventory:{product_id}"
        with self.lock(lock_key, timeout=5) as acquired:
            if not acquired:
                return False
            try:
                # 查询当前库存
                current = int(self.redis.get(f"stock:{product_id}") or 0)
                if current < quantity:
                    print(f"库存不足，当前库存: {current}, 需要: {quantity}")
                    return False
                # 扣减库存
                new_stock = current - quantity
                self.redis.set(f"stock:{product_id}", new_stock)
                print(f"扣减库存成功，剩余: {new_stock}")
                return True
            except Exception as e:
                print(f"扣减库存失败: {e}")
                return False
# 使用示例
def business_example():
    lock_service = DistributedLockService()
    # 场景1：处理订单
    try:
        lock_service.process_order("order_001", "user_123", 99.99)
    except Exception as e:
        print(f"订单处理异常: {e}")
    # 场景2：库存扣减
    success = lock_service.inventory_deduction("product_001", 10)
    if success:
        print("库存扣减成功")
    else:
        print("库存扣减失败")
# 启动多个进程模拟并发
from multiprocessing import Process
def simulate_concurrent_orders(order_id):
    lock_service = DistributedLockService()
    try:
        lock_service.process_order(order_id, "user_proc", 100)
    except Exception as e:
        print(f"进程处理失败: {e}")
def test_concurrent():
    processes = []
    for i in range(3):
        p = Process(target=simulate_concurrent_orders, args=(f"order_test_001",))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
if __name__ == "__main__":
    # 先确保Redis运行
    try:
        # 运行示例
        business_example()
        # 测试并发场景
        # test_concurrent()
    except Exception as e:
        print(f"错误: {e}")

使用现成的库：redlock-py

from redlock import RedLock
import time
def use_redlock_lib():
    # 配置Redis服务器
    servers = [
        {"host": "localhost", "port": 6379, "db": 0},
        {"host": "localhost", "port": 6380, "db": 0},
        {"host": "localhost", "port": 6381, "db": 0}
    ]
    # 使用RedLock
    lock = RedLock(
        "my_resource_lock",
        connection_details=servers,
        ttl=10000  # 10秒
    )
    if lock.acquire():
        try:
            print("获取锁成功")
            # 执行业务逻辑
            time.sleep(2)
        finally:
            lock.release()
    # 或者使用上下文管理器
    with RedLock("resource_lock", connection_details=servers, ttl=10000):
        print("在锁保护下执行操作")
        time.sleep(1)
  1. 原子性：使用SET NX EX或Lua脚本确保原子操作
  2. 唯一标识：每个客户端生成唯一value，防止误删其他客户端的锁
  3. 过期时间：设置合理超时防止死锁
  4. 重试机制：获取失败时支持重试，使用退避策略
  5. 锁续期：长时间业务可以自动续期

选择哪种实现方式取决于你的具体需求：

  • 简单场景：基础SETNX实现
  • 需要重试：带重试机制的锁
  • 高可用要求：RedLock算法
  • 快速开发：使用现成的redlock-py库

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇Python数据分析案例如何清洗数据

下一篇Python Redis列表案例如何操作列表数据

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝