Database-Based Job Locking — Complete Guide
In this tutorial, you will learn about Database. We cover key concepts, practical examples, and best practices to help you master this topic.
Use database-level locks for job coordination including advisory locks, row-level locks, SELECT FOR UPDATE, and optimistic locking for background processing.
What You Learn
You will learn how to use PostgreSQL advisory locks, MySQL GET_LOCK, row-level locks with SELECT FOR UPDATE, and optimistic locking with version columns for job coordination.
Why It Matters
Database locks are available without external dependencies like Redis. They are transactional, meaning locks are automatically released on Transaction commit or rollback, reducing orphaned locks.
Real-World Use
DodaTech uses PostgreSQL advisory locks for coordinating recurring job execution across multiple workers. The lock is acquired at the start of the job transaction and automatically released on completion.
PostgreSQL Advisory Lock
import psycopg2
import time
import threading
class AdvisoryLock:
def __init__(self, conn):
self.conn = conn
def acquire(self, lock_id, blocking=True):
lock_id = abs(hash(lock_id)) % (2**31)
if blocking:
self.conn.cursor().execute("SELECT pg_advisory_lock(%s)", (lock_id,))
return True
else:
cursor = self.conn.cursor()
cursor.execute("SELECT pg_try_advisory_lock(%s)", (lock_id,))
return cursor.fetchone()[0]
def release(self, lock_id):
lock_id = abs(hash(lock_id)) % (2**31)
self.conn.cursor().execute("SELECT pg_advisory_unlock(%s)", (lock_id,))
self.conn.commit()
# Simulated PostgreSQL advisory lock
class SimulatedAdvisoryLock:
def __init__(self):
self._locks = set()
self._lock = threading.Lock()
def acquire(self, lock_name):
with self._lock:
if lock_name in self._locks:
return False
self._locks.add(lock_name)
return True
def release(self, lock_name):
with self._lock:
self._locks.discard(lock_name)
def is_locked(self, lock_name):
with self._lock:
return lock_name in self._locks
s_lock = SimulatedAdvisoryLock()
print(f"Acquire backup: {s_lock.acquire('backup_lock')}")
print(f"Acquire backup again: {s_lock.acquire('backup_lock')}")
s_lock.release('backup_lock')
print(f"After release: {s_lock.acquire('backup_lock')}")
Expected output:
Acquire backup: True
Acquire backup again: False
After release: True
SELECT FOR UPDATE Lock
import sqlite3
import time
import threading
class RowLevelLock:
def __init__(self, db_path=':memory:'):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.execute('''
CREATE TABLE IF NOT EXISTS job_locks (
job_id TEXT PRIMARY KEY,
status TEXT,
locked_by TEXT,
locked_at REAL,
version INTEGER DEFAULT 1
)
''')
self.conn.commit()
def acquire_job(self, job_id, worker_id):
cursor = self.conn.execute(
'SELECT status, version FROM job_locks WHERE job_id = ?',
(job_id,)
)
row = cursor.fetchone()
if row:
status, version = row
if status == 'locked':
locked_cursor = self.conn.execute(
'SELECT locked_at FROM job_locks WHERE job_id = ?',
(job_id,)
)
locked_row = locked_cursor.fetchone()
if locked_row and time.time() - locked_row[0] < 30:
return False
self.conn.execute(
'''INSERT OR REPLACE INTO job_locks (job_id, status, locked_by, locked_at, version)
VALUES (?, ?, ?, ?, COALESCE((SELECT version + 1 FROM job_locks WHERE job_id = ?), 1))''',
(job_id, 'locked', worker_id, time.time(), job_id)
)
self.conn.commit()
return True
def release_job(self, job_id, worker_id):
self.conn.execute(
'''UPDATE job_locks SET status = 'completed' WHERE job_id = ? AND locked_by = ?''',
(job_id, worker_id)
)
self.conn.commit()
def get_status(self, job_id):
cursor = self.conn.execute(
'SELECT status, locked_by FROM job_locks WHERE job_id = ?',
(job_id,)
)
row = cursor.fetchone()
if row:
return {'status': row[0], 'locked_by': row[1]}
return None
db_lock = RowLevelLock()
def worker_process(job_id, worker_id):
if db_lock.acquire_job(job_id, worker_id):
print(f"[{worker_id}] Processing {job_id}")
time.sleep(1)
db_lock.release_job(job_id, worker_id)
print(f"[{worker_id}] Released {job_id}")
else:
print(f"[{worker_id}] Could not acquire {job_id}")
t1 = threading.Thread(target=worker_process, args=('job-1', 'w1'), daemon=True)
t2 = threading.Thread(target=worker_process, args=('job-1', 'w2'), daemon=True)
t1.start()
t2.start()
t1.join(timeout=3)
t2.join(timeout=3)
Expected output:
[w1] Processing job-1
[w2] Could not acquire job-1
[w1] Released job-1
Optimistic Locking
import sqlite3
import time
import threading
class OptimisticLock:
def __init__(self, db_path=':memory:'):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.execute('''
CREATE TABLE IF NOT EXISTS job_processing (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_type TEXT,
status TEXT,
version INTEGER DEFAULT 1,
updated_at REAL
)
''')
self.conn.execute(
"INSERT INTO job_processing (job_type, status, version, updated_at) VALUES (?, ?, ?, ?)",
('email_campaign', 'pending', 1, time.time())
)
self.conn.commit()
def try_process(self, job_id, worker_id):
while True:
cursor = self.conn.execute(
'SELECT id, status, version FROM job_processing WHERE id = ?',
(job_id,)
)
row = cursor.fetchone()
if not row:
return False
_, status, version = row
if status != 'pending':
return False
self.conn.execute(
'''UPDATE job_processing SET status = ?, version = ?, updated_at = ?
WHERE id = ? AND version = ?''',
('processing', version + 1, time.time(), job_id, version)
)
self.conn.commit()
if self.conn.total_changes > 0:
print(f"[{worker_id}] Acquired job {job_id} (v{version})")
return True
time.sleep(0.05)
def complete(self, job_id, worker_id):
self.conn.execute(
'''UPDATE job_processing SET status = 'completed', updated_at = ? WHERE id = ?''',
(time.time(), job_id)
)
self.conn.commit()
print(f"[{worker_id}] Completed job {job_id}")
opt = OptimisticLock()
def worker(wid):
if opt.try_process(1, wid):
time.sleep(0.5)
opt.complete(1, wid)
threads = [threading.Thread(target=worker, args=(f'w{i}',), daemon=True) for i in range(3)]
for t in threads:
t.start()
for t in threads:
t.join(timeout=2)
Expected output:
[w0] Acquired job 1 (v1)
[w0] Completed job 1
MySQL GET_LOCK
import time
import threading
class MySQLGetLockSimulator:
def __init__(self):
self._locks = {}
self._lock = threading.Lock()
def get_lock(self, lock_name, timeout=10):
with self._lock:
if lock_name in self._locks:
return False
self._locks[lock_name] = threading.current_thread().ident
return True
def release_lock(self, lock_name):
with self._lock:
if self._locks.get(lock_name) == threading.current_thread().ident:
del self._locks[lock_name]
return 1
return 0
def is_free_lock(self, lock_name):
with self._lock:
return lock_name not in self._locks
mysql_lock = MySQLGetLockSimulator()
def named_lock_worker(wid, lock_name):
if mysql_lock.get_lock(lock_name):
print(f"[{wid}] Got lock: {lock_name}")
time.sleep(1)
mysql_lock.release_lock(lock_name)
print(f"[{wid}] Released: {lock_name}")
else:
print(f"[{wid}] Lock busy: {lock_name}")
threads = [
threading.Thread(target=named_lock_worker, args=('w1', 'etl'), daemon=True),
threading.Thread(target=named_lock_worker, args=('w2', 'etl'), daemon=True),
]
for t in threads:
t.start()
for t in threads:
t.join(timeout=3)
Expected output:
[w1] Got lock: etl
[w2] Lock busy: etl
[w1] Released: etl
Common Mistakes
1. Not Releasing Locks on Error
Unhandled exceptions skip lock release. Use try/finally or context managers to ensure release.
2. Lock Contention with Long Operations
Holding database locks for slow I/O operations blocks other workers. Do the I/O outside the lock, only lock for critical sections.
3. Deadlocks with Multiple Locks
Acquiring locks in different orders causes deadlocks. Always acquire locks in a consistent order across all workers.
4. Using Row Locks for Non-Transactional Operations
SELECT FOR UPDATE only works within transactions. Without transactions, the lock is immediately released.
5. Assuming Database Locks Are Free
Database locks consume connections and memory. Monitor lock contention and optimize lock duration.
Practice Questions
1. What is the difference between advisory locks and row-level locks?
Advisory locks are application-defined names. Row-level locks lock specific rows in a table. Advisory locks are lighter weight.
2. How does optimistic locking work?
Read the version number, attempt update where version matches, retry if version changed. No locks needed, but retries may be needed.
3. When should you use database locks over Redis locks?
When your application already uses a database, database locks simplify the stack. They also integrate with transactions.
4. What is a Deadlock and how to prevent it?
Two workers each holding a lock the other needs. Prevent by acquiring locks in a consistent global order.
Challenge
Build a job locking system using database locks: PostgreSQL advisory locks for job coordination, row-level locks for resource access, optimistic locking for contention-heavy jobs, and deadlock detection with retry.
FAQ
Mini Project: Database Lock Manager
import sqlite3
import time
import threading
class DatabaseLockManager:
def __init__(self):
self.conn = sqlite3.connect(':memory:')
self.conn.execute('CREATE TABLE IF NOT EXISTS locks (name TEXT PRIMARY KEY, holder TEXT, acquired REAL)')
def acquire(self, name, holder, ttl=30):
now = time.time()
self.conn.execute('DELETE FROM locks WHERE acquired < ?', (now - ttl,))
try:
self.conn.execute(
'INSERT INTO locks (name, holder, acquired) VALUES (?, ?, ?)',
(name, holder, now)
)
self.conn.commit()
return True
except sqlite3.IntegrityError:
return False
def release(self, name, holder):
self.conn.execute(
'DELETE FROM locks WHERE name = ? AND holder = ?',
(name, holder)
)
self.conn.commit()
mgr = DatabaseLockManager()
print(mgr.acquire('job-1', 'w1'))
print(mgr.acquire('job-1', 'w2'))
mgr.release('job-1', 'w1')
print(mgr.acquire('job-1', 'w2'))
Expected output:
True
False
True
What's Next
Now that you understand database locking, explore cron timezone handling for timezone-aware scheduling, then learn about job monitoring alerting.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro