from contextlib import contextmanager @contextmanager def lock(redis, lock_key, expire): try: locked = redis.set(lock_key, 'locked', expire) yield locked finally: redis.delete(lock_key) # 业务调用 with 代码块执行结束后 自动释放锁资源 with lock(redis, 'locked', 3) as locked: if not locked: return # do something ...