1"""Chapter 22: Message queues. Tested with Python 3, standard library only.
2
3Running example: the grocery app. When an order is placed, the checkout service puts a
4job on a queue; workers send the SMS, print the packing slip and add loyalty points.
5"""
6import heapq
7import zlib
8import random
9from collections import defaultdict, deque
10
11
12# ---------- 22.4 / 22.5 a queue with acks, visibility timeouts and a DLQ ----------
13
14class Queue:
15 def __init__(self, visibility_timeout=30, max_receives=3):
16 self.visibility_timeout = visibility_timeout
17 self.max_receives = max_receives
18 self.messages = {} # id -> body
19 self.invisible_until = {} # id -> time
20 self.receives = defaultdict(int)
21 self.order = deque()
22 self.dlq = []
23 self.next_id = 1
24
25 def send(self, body):
26 mid = self.next_id
27 self.next_id += 1
28 self.messages[mid] = body
29 self.order.append(mid)
30 return mid
31
32 def receive(self, now):
33 for mid in list(self.order):
34 if mid not in self.messages:
35 continue
36 if self.invisible_until.get(mid, -1) > now:
37 continue
38 if self.receives[mid] >= self.max_receives: # poison message
39 self.dlq.append(self.messages.pop(mid))
40 continue
41 self.receives[mid] += 1
42 self.invisible_until[mid] = now + self.visibility_timeout
43 return mid, self.messages[mid]
44 return None
45
46 def ack(self, mid):
47 self.messages.pop(mid, None)
48
49 def depth(self):
50 return len(self.messages)
51
52
53# ---------- 22.3 delivery guarantees ----------
54
55def deliver(mode, crash_points, n=100):
56 """Process n messages; crash_points: message numbers where the worker crashes.
57 'before' = crash after ack but before processing; 'after' = after processing, before ack."""
58 q = Queue(visibility_timeout=30)
59 for i in range(n):
60 q.send(i)
61 processed = []
62 now = 0
63 while q.depth():
64 got = q.receive(now)
65 if got is None:
66 now += 31 # wait for invisible messages to return
67 continue
68 mid, body = got
69 crash = crash_points.get(body)
70 if mode == "at-most-once":
71 q.ack(mid) # ack first
72 if crash == "before":
73 crash_points.pop(body)
74 continue # lost
75 processed.append(body)
76 else: # at-least-once
77 processed.append(body) # process first
78 if crash == "after":
79 crash_points.pop(body)
80 continue # no ack: it comes back
81 q.ack(mid)
82 return processed
83
84
85# ---------- 22.6 idempotent consumer ----------
86
87class Loyalty:
88 def __init__(self):
89 self.points = defaultdict(int)
90 self.applied = set() # message / order ids already applied
91
92 def award_naive(self, order_id, user, pts):
93 self.points[user] += pts
94
95 def award_idempotent(self, order_id, user, pts):
96 if order_id in self.applied: # in real life: a unique constraint
97 return False
98 self.applied.add(order_id)
99 self.points[user] += pts
100 return True
101
102
103# ---------- 22.7 ordering ----------
104
105def process_parallel(events, workers, seed=22):
106 """Events for many orders, handed round-robin to workers with random speed."""
107 rnd = random.Random(seed)
108 finished = []
109 for i, ev in enumerate(events):
110 finished.append((i // workers + rnd.random() * 1.5, ev))
111 return [ev for _, ev in sorted(finished)]
112
113
114def process_partitioned(events, partitions):
115 """Same key -> same partition -> one consumer, in order."""
116 parts = defaultdict(list)
117 for ev in events:
118 parts[zlib.crc32(ev[0].encode()) % partitions].append(ev)
119 out = []
120 for p in sorted(parts):
121 out.extend(parts[p])
122 return out
123
124
125def in_order_per_key(events, steps):
126 pos = defaultdict(int)
127 for key, step in events:
128 if steps.index(step) < pos[key]:
129 return False
130 pos[key] = steps.index(step)
131 return True
132
133
134# ---------- 22.8 backpressure ----------
135
136def backlog(in_rate, out_rate, seconds):
137 return max(0, (in_rate - out_rate) * seconds)
138
139
140def drain_time(backlog_msgs, out_rate, in_rate):
141 return backlog_msgs / (out_rate - in_rate)
142
143
144# ---------- 22.9 scaling consumers ----------
145
146def workers_needed(rate_per_s, seconds_per_msg, headroom=1.0):
147 return int(-(-rate_per_s * seconds_per_msg * headroom // 1)) # ceiling
148
149
150# ---------- 22.10 log vs queue ----------
151
152class Log:
153 """Kafka-style: messages stay; each consumer group keeps its own offset."""
154
155 def __init__(self):
156 self.entries, self.offsets = [], defaultdict(int)
157
158 def append(self, msg):
159 self.entries.append(msg)
160
161 def poll(self, group, max_n=10):
162 start = self.offsets[group]
163 batch = self.entries[start:start + max_n]
164 self.offsets[group] = start + len(batch)
165 return batch
166
167 def seek(self, group, offset):
168 self.offsets[group] = offset
169
170
171# ---------- 22.11 delayed jobs ----------
172
173class Scheduler:
174 def __init__(self):
175 self.heap = []
176
177 def schedule(self, run_at, job):
178 heapq.heappush(self.heap, (run_at, job))
179
180 def due(self, now):
181 out = []
182 while self.heap and self.heap[0][0] <= now:
183 out.append(heapq.heappop(self.heap)[1])
184 return out
185
186
187def backoff_schedule(attempts, base=2, cap=300):
188 return [min(cap, base ** n) for n in range(1, attempts + 1)]
189
190
191# ---------- checks ----------
192
193def run_checks():
194 # visibility timeout: an un-acked message comes back after 30 s
195 q = Queue(visibility_timeout=30)
196 q.send("sms for order 881")
197 mid, _ = q.receive(now=0)
198 assert q.receive(now=10) is None # hidden while a worker has it
199 assert q.receive(now=31)[0] == mid # worker died: it reappears
200 q.ack(mid)
201 assert q.depth() == 0
202
203 # poison message goes to the DLQ after 3 tries
204 q = Queue(visibility_timeout=30, max_receives=3)
205 q.send("order 999 with a malformed address")
206 t = 0
207 for _ in range(3):
208 assert q.receive(t) is not None # fails each time, never acked
209 t += 31
210 assert q.receive(t) is None and q.dlq == ["order 999 with a malformed address"]
211
212 # at-most-once loses, at-least-once duplicates
213 amo = deliver("at-most-once", {7: "before", 42: "before"})
214 alo = deliver("at-least-once", {7: "after", 42: "after"})
215 assert len(amo) == 98 and 7 not in amo # two messages lost
216 assert len(alo) == 102 and alo.count(7) == 2 # two messages processed twice
217 assert set(alo) == set(range(100))
218
219 # idempotent consumer: duplicates don't double the points
220 naive, safe = Loyalty(), Loyalty()
221 deliveries = [("881", "asha", 50), ("882", "ravi", 30), ("881", "asha", 50)] # 881 redelivered
222 for d in deliveries:
223 naive.award_naive(*d)
224 safe.award_idempotent(*d)
225 assert naive.points["asha"] == 100 and safe.points["asha"] == 50
226
227 # ordering: parallel workers scramble steps; partitioning by order id keeps them
228 steps = ["placed", "paid", "packed", "shipped"]
229 events = [(f"order{o}", s) for o in range(20) for s in steps] # each order's steps, in order
230 scrambled = process_parallel(events, workers=8)
231 assert not in_order_per_key(scrambled, steps)
232 assert in_order_per_key(process_partitioned(events, partitions=6), steps)
233
234 # backpressure: Big Sale peak, 1,200 orders/s in, workers do 1,000/s
235 b = backlog(1_200, 1_000, 3_600)
236 assert b == 720_000 # one hour of peak
237 assert drain_time(b, out_rate=1_000, in_rate=400) == 1_200 # 20 minutes after peak
238
239 # scaling consumers: 1,000 msgs/s at 200 ms each
240 assert workers_needed(1_000, 0.2) == 200
241 assert workers_needed(1_000, 0.2, headroom=1.5) == 300
242
243 # log vs queue: two groups read the same events independently, and one replays
244 log = Log()
245 for i in range(5):
246 log.append(f"order{i} placed")
247 assert log.poll("sms") == log.poll("analytics") # both see all five
248 log.seek("analytics", 0)
249 assert len(log.poll("analytics")) == 5 # replay
250 assert log.poll("sms") == []
251
252 # delayed jobs: cancel unpaid orders after 15 minutes
253 s = Scheduler()
254 s.schedule(900, "cancel order 881 if unpaid")
255 s.schedule(60, "reminder: complete payment")
256 assert s.due(59) == [] and s.due(60) == ["reminder: complete payment"]
257 assert s.due(900) == ["cancel order 881 if unpaid"]
258 assert backoff_schedule(6) == [2, 4, 8, 16, 32, 64]
259
260 # Level 4: visibility timeout shorter than processing time -> double processing
261 q = Queue(visibility_timeout=30)
262 q.send(("881", "asha", 50))
263 bug = Loyalty()
264 m1 = q.receive(now=0) # worker A starts, takes 45 s
265 m2 = q.receive(now=31) # reappears: worker B takes it too
266 assert m1[0] == m2[0]
267 bug.award_naive(*m1[1])
268 bug.award_naive(*m2[1])
269 assert bug.points["asha"] == 100
270 fixed = Loyalty()
271 fixed.award_idempotent(*m1[1])
272 fixed.award_idempotent(*m2[1])
273 assert fixed.points["asha"] == 50
274
275 return {"amo": len(amo), "alo": len(alo), "backlog": b, "workers": 200}
276
277
278if __name__ == "__main__":
279 print("All checks pass.", run_checks())