Python实现Redis分布式锁的示例代码与解析

一颗松果

发布时间：2026-10-13| 5人浏览过

原创

本文介绍基于Python和Redis实现的分布式锁功能，通过setnx命令与时间戳机制解决并发冲突。示例代码包含获取锁、释放锁及装饰器用法，适用于多进程或分布式环境下的资源同步场景。

本文通过一个具体的Python实例，详细讲解了如何利用Redis实现分布式锁功能。该方案旨在为开发者提供可参考的实现思路，帮助理解在分布式环境中进行资源同步的基本方法。

#!/usr/bin/env python
# coding=utf-8
import time
import redis
class RedisLock(object):
  def __init__(self, key):
    self.rdcon = redis.Redis(host='', port=6379, password="", db=1)
    self._lock = 0
    self.lock_key = "%s_dynamic_test" % key
  @staticmethod
  def get_lock(cls, timeout=10):
    while cls._lock != 1:
      timestamp = time.time() + timeout + 1
      cls._lock = cls.rdcon.setnx(cls.lock_key, timestamp)
      if cls._lock == 1 or (time.time() > cls.rdcon.get(cls.lock_key) and time.time() > cls.rdcon.getset(cls.lock_key, timestamp)):
        print "get lock"
        break
      else:
        time.sleep(0.3)
  @staticmethod
  def release(cls):
    if time.time() < cls.rdcon.get(cls.lock_key):
      print "release lock"
      cls.rdcon.delete(cls.lock_key)
def deco(cls):
  def _deco(func):
    def __deco(*args, **kwargs):
      print "before %s called [%s]."%(func.__name__, cls)
      cls.get_lock(cls)
      try:
        return func(*args, **kwargs)
      finally:
        cls.release(cls)
    return __deco
  return _deco
@deco(RedisLock("112233"))
def myfunc():
  print "myfunc() called."
  time.sleep(20)
if __name__ == "__main__":
  myfunc()

如果您对Python的其他实用技巧感兴趣，建议进一步查阅本站的相关专题内容。这些专题涵盖了数据库操作、编码处理、数据结构与算法、Socket编程、函数使用、字符串处理、基础入门进阶以及文件目录操作等多个方面，能够为您提供更全面的技术支持。

希望上述关于Python程序设计的讲解与示例，能对您的实际开发工作提供有效的帮助。