Pythontask_queue.py
Copy
1
import json
2
import time
3
import uuid
4
import random
5
import threading
6
import redis
7
 
8
# docker run -d -p 6379:6379 redis:alpine
9
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
10
 
11
QUEUE_KEY = "tasks:queue"
12
PROCESSING_KEY = "tasks:processing"
13
DLQ_KEY = "tasks:dead-letter"
14
VISIBILITY_TIMEOUT = 30  # seconds
15
 
16
# ── Task serialization ────────────────────────────────────────────────
17
def make_task(task_type: str, payload: dict, max_retries: int = 3) -> dict:
18
    return {
19
        "id": str(uuid.uuid4()),
20
        "type": task_type,
21
        "payload": payload,
22
        "attempts": 0,
23
        "max_retries": max_retries,
24
        "created_at": time.time(),
25
        "enqueued_at": time.time(),
26
    }
27
 
28
# ── Producer ──────────────────────────────────────────────────────────
29
def enqueue(task: dict) -> str:
30
    r.lpush(QUEUE_KEY, json.dumps(task))
31
    return task["id"]
32
 
33
# ── At-least-once delivery with visibility timeout ───────────────────
34
def dequeue(timeout: int = 5) -> dict | None:
35
    """Atomically move task from queue to processing set."""
36
    item = r.brpop(QUEUE_KEY, timeout=timeout)
37
    if not item:
38
        return None
39
    
40
    _, raw = item
41
    task = json.loads(raw)
42
    task["attempts"] += 1
43
    task["visible_after"] = time.time() + VISIBILITY_TIMEOUT
44
    
45
    # Store in processing set with score = visibility timeout
46
    r.zadd(PROCESSING_KEY, {json.dumps(task): task["visible_after"]})
47
    return task
48
 
49
def ack(task: dict):
50
    """Acknowledge successful processing — remove from processing set."""
51
    r.zrem(PROCESSING_KEY, json.dumps(task))
52
 
53
def nack(task: dict, reason: str = ""):
54
    """Negative acknowledgment — requeue or move to DLQ."""
55
    r.zrem(PROCESSING_KEY, json.dumps(task))
56
    
57
    task["last_error"] = reason
58
    if task["attempts"] < task["max_retries"]:
59
        # Exponential backoff: delay requeue
60
        delay = 2 ** task["attempts"]
61
        task["enqueued_at"] = time.time()
62
        time.sleep(delay * 0.1)  # Scaled for demo
63
        r.lpush(QUEUE_KEY, json.dumps(task))
64
        print(f"  ↩ Requeued task {task['id'][:8]} (attempt {task['attempts']}/{task['max_retries']})")
65
    else:
66
        # Dead-letter
67
        r.lpush(DLQ_KEY, json.dumps(task))
68
        print(f"  ☠ Task {task['id'][:8]} moved to DLQ after {task['attempts']} attempts")
69
 
70
def requeue_invisible_tasks():
71
    """Requeue tasks whose visibility timeout has expired (for crashed workers)."""
72
    now = time.time()
73
    expired = r.zrangebyscore(PROCESSING_KEY, 0, now, withscores=True)
74
    for raw, _ in expired:
75
        r.zrem(PROCESSING_KEY, raw)
76
        task = json.loads(raw)
77
        r.lpush(QUEUE_KEY, json.dumps(task))
78
        print(f"  ♻ Requeued expired task {task['id'][:8]}")
79
 
80
# ── Worker ────────────────────────────────────────────────────────────
81
def process_task(task: dict) -> bool:
82
    """Simulate task processing. Returns True on success."""
83
    time.sleep(random.uniform(0.01, 0.05))
84
    # 30% failure rate for demo
85
    return random.random() > 0.3
86
 
87
processed_count = 0
88
failed_count = 0
89
 
90
def worker(worker_id: str, stop_event: threading.Event):
91
    global processed_count, failed_count
92
    while not stop_event.is_set():
93
        task = dequeue(timeout=1)
94
        if not task:
95
            continue
96
        
97
        try:
98
            success = process_task(task)
99
            if success:
100
                ack(task)
101
                processed_count += 1
102
                print(f"  [{worker_id}] ✓ {task['type']} {task['id'][:8]}")
103
            else:
104
                nack(task, reason="processing_failed")
105
                failed_count += 1
106
        except Exception as e:
107
            nack(task, reason=str(e))
108
 
109
# ── Run demo ──────────────────────────────────────────────────────────
110
# Flush queues
111
for key in [QUEUE_KEY, PROCESSING_KEY, DLQ_KEY]:
112
    r.delete(key)
113
 
114
# Enqueue 20 tasks
115
task_types = ["send_email", "resize_image", "generate_report", "sync_data"]
116
for i in range(20):
117
    task = make_task(random.choice(task_types), {"item_id": i, "data": f"payload-{i}"})
118
    enqueue(task)
119
print(f"Enqueued 20 tasks\n")
120
 
121
# Start 3 workers
122
stop = threading.Event()
123
workers = [threading.Thread(target=worker, args=(f"W{i}", stop)) for i in range(3)]
124
for w in workers:
125
    w.start()
126
 
127
time.sleep(3)
128
stop.set()
129
for w in workers:
130
    w.join(timeout=2)
131
 
132
print(f"\n=== Results ===")
133
print(f"Processed: {processed_count}")
134
print(f"Failed: {r.llen(DLQ_KEY)} in DLQ")
135
print(f"Remaining queue: {r.llen(QUEUE_KEY)}")
136