Distributed Locks Complete Guide 2025: Redlock, Zookeeper, etcd, Fencing Token, Safe Implementation

lock = threading.Lock()

with lock:
    critical_section()

[Server A] ←─ network ─→ [Lock Service] ←─ network ─→ [Server B]

def acquire_lock(key, ttl=10):
    return redis.set(key, "locked", nx=True, ex=ttl)

def release_lock(key):
    redis.delete(key)

# Usage
if acquire_lock("my-resource"):
    try:
        do_work()
    finally:
        release_lock("my-resource")

# Client A acquires the lock
acquire_lock("my-resource")
# Client A is working...
# Work takes 30 seconds (exceeds 10s TTL)
# Lock expires automatically

# Client B acquires the lock
acquire_lock("my-resource")

# Client A finishes and releases the lock
release_lock("my-resource")
# → Releases Client B's lock!

# Client C can now acquire the lock
# Client B and C work concurrently!

import uuid

def acquire_lock(key, ttl=10):
    token = str(uuid.uuid4())
    if redis.set(key, token, nx=True, ex=ttl):
        return token
    return None

def release_lock(key, token):
    # Atomicity via Lua script
    script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    redis.eval(script, 1, key, token)

[Redis] ← if it dies, all locks are lost
   ↓
[Server A] [Server B] [Server C]

[Master Redis]
    ↓ async replication
[Slave Redis]

[Redis 1] [Redis 2] [Redis 3] [Redis 4] [Redis 5]
  ↓        ↓         ↓         ↓         ↓
[Client] → try SET NX EX on all
        → 3/5 or more successes = lock acquired

def acquire_redlock(key, ttl=10000):
    token = str(uuid.uuid4())
    quorum = len(redis_clients) // 2 + 1  # 3/5
    
    start_time = time.time() * 1000
    acquired = []
    
    for client in redis_clients:
        try:
            # Very short timeout
            if client.set(key, token, nx=True, px=ttl):
                acquired.append(client)
        except:
            pass
    
    elapsed = time.time() * 1000 - start_time
    
    if len(acquired) >= quorum and elapsed < ttl:
        return token  # Lock acquired
    else:
        # Failure - release everything
        for client in acquired:
            release_lock(client, key, token)
        return None

1. Client A acquires lock (10s TTL)
2. Client A stalls 15s on GC (lock expires)
3. Client B acquires lock
4. Client A wakes up → thinks it still holds the lock
5. Client A and B work concurrently

1. Client A acquires lock (10s TTL)
2. Redis clock suddenly jumps 5s (NTP sync)
3. Lock expires early
4. Client B acquires lock

1. Client A acquires lock, token = 33
2. Client A stalls on GC
3. Client B acquires lock, token = 34
4. Client B writes to storage (token 34)
5. Client A wakes up, writes to storage (token 33)
6. Storage sees 33 < 34 → rejects

def acquire_lock_with_token(key):
    token = redis.incr(f"{key}:token")  # Monotonic
    if redis.set(key, token, nx=True, ex=10):
        return token
    return None

def write_to_storage(data, token):
    # Storage validates the token
    if storage.last_token >= token:
        raise StaleTokenError()
    storage.write(data)
    storage.last_token = token

def acquire_lock(zk, lock_path):
    # 1. Create sequential ephemeral node
    my_node = zk.create(
        f"{lock_path}/lock-",
        ephemeral=True,
        sequence=True
    )
    # e.g. /lock/lock-0000000005
    
    while True:
        # 2. Get all child nodes
        children = sorted(zk.get_children(lock_path))
        
        # 3. If I'm smallest, acquire the lock
        if my_node.endswith(children[0]):
            return my_node
        
        # 4. Otherwise watch the node right before me
        my_index = children.index(my_node.split('/')[-1])
        prev_node = children[my_index - 1]
        
        zk.exists(f"{lock_path}/{prev_node}", watch=callback)
        wait_for_callback()

import org.apache.curator.framework.recipes.locks.InterProcessMutex;

InterProcessMutex lock = new InterProcessMutex(client, "/my-lock");
if (lock.acquire(10, TimeUnit.SECONDS)) {
    try {
        // work
    } finally {
        lock.release();
    }
}

import etcd3

client = etcd3.client()

# Create lease (10 seconds)
lease = client.lease(ttl=10)

# Acquire lock
lock = client.lock("my-lock", ttl=10)
lock.acquire()

try:
    do_work()
finally:
    lock.release()

BEGIN;
SELECT * FROM accounts WHERE id = 123 FOR UPDATE;
-- Hold the lock, other transactions wait
UPDATE accounts SET balance = balance - 100 WHERE id = 123;
COMMIT;
-- Release the lock

-- Acquire lock
SELECT pg_advisory_lock(12345);

-- Work
UPDATE ...;

-- Release lock
SELECT pg_advisory_unlock(12345);

-- 1. Read data and version
SELECT *, version FROM users WHERE id = 1;
-- version = 5

-- 2. Update after the work
UPDATE users SET name = 'New', version = version + 1
WHERE id = 1 AND version = 5;
-- If affected rows = 0, fail (someone else modified first)

import threading

class RenewingLock:
    def __init__(self, key, ttl=10):
        self.key = key
        self.token = None
        self.ttl = ttl
        self.stop_renewal = False
    
    def acquire(self):
        self.token = acquire_lock(self.key, self.ttl)
        if self.token:
            self.renewal_thread = threading.Thread(target=self._renew)
            self.renewal_thread.start()
        return self.token
    
    def _renew(self):
        while not self.stop_renewal:
            time.sleep(self.ttl / 3)
            redis.expire(self.key, self.ttl)
    
    def release(self):
        self.stop_renewal = True
        release_lock(self.key, self.token)

import random

def acquire_with_retry(key, max_retries=10):
    for attempt in range(max_retries):
        if token := acquire_lock(key):
            return token
        # Exponential backoff + jitter
        delay = (2 ** attempt) * 0.1 + random.uniform(0, 0.5)
        time.sleep(delay)
    return None

class ReentrantLock:
    def __init__(self, key):
        self.key = key
        self.count = 0
        self.token = None
    
    def acquire(self):
        if self.count > 0:
            self.count += 1
            return True
        
        self.token = acquire_lock(self.key)
        if self.token:
            self.count = 1
            return True
        return False
    
    def release(self):
        self.count -= 1
        if self.count == 0:
            release_lock(self.key, self.token)
            self.token = None

# Wrong - deadlock possible
def transfer(from_id, to_id, amount):
    lock1 = acquire_lock(f"account:{from_id}")
    lock2 = acquire_lock(f"account:{to_id}")
    # ...

# Two users simultaneously transferring to each other:
# A → B: lock A, lock B
# B → A: lock B, lock A
# → Deadlock

# Correct - sorted ordering
def transfer(from_id, to_id, amount):
    first, second = sorted([from_id, to_id])
    lock1 = acquire_lock(f"account:{first}")
    lock2 = acquire_lock(f"account:{second}")
    # ...

[Job Queue]
    ↓
[Worker 1] (jobs A, B)
[Worker 2] (jobs C, D)
[Worker 3] (jobs E, F)

def process_payment(payment_id, amount):
    if db.exists(f"processed:{payment_id}"):
        return  # Already processed
    
    db.atomic_set(f"processed:{payment_id}", True)
    actually_charge(amount)

UPDATE inventory 
SET quantity = quantity - 1 
WHERE product_id = 123 AND quantity > 0;

events.append(OrderCreated(...))
events.append(InventoryReserved(...))
events.append(PaymentCharged(...))

class AccountActor extends Actor {
  var balance: BigDecimal = 0
  
  def receive = {
    case Deposit(amount) => balance += amount
    case Withdraw(amount) => balance -= amount
  }
}

Do you really need a distributed lock?
├─ No → use idempotency, CAS, partitioning
└─ Yes
   ├─ Efficiency goal (roughly OK)
   │  └─ Single Redis lock (simple)
   └─ Correctness goal (never OK)
      ├─ ZooKeeper / etcd
      └─ Fencing token (if storage supports)