1"""Chapter 23: Pub/sub and fan-out. Tested with Python 3, standard library only.
2
3Running example: the bookstore's "follow an author" feature. When an author releases a
4book, every follower should see it in "New from authors you follow". Plus the
5"order placed" event that several services react to.
6"""
7import random
8from collections import defaultdict
9
10
11# ---------- 23.1-23.3 topics, subscribers, consumer groups ----------
12
13class Broker:
14 """Each subscription gets its own copy of every message on the topic.
15 Inside a subscription (consumer group), each message goes to one member."""
16
17 def __init__(self):
18 self.subs = defaultdict(dict) # topic -> {group: [members]}
19 self.inbox = defaultdict(list) # member -> messages
20 self.rr = defaultdict(int)
21
22 def subscribe(self, topic, group, member):
23 self.subs[topic].setdefault(group, []).append(member)
24
25 def publish(self, topic, msg):
26 for group, members in self.subs[topic].items():
27 m = members[self.rr[(topic, group)] % len(members)]
28 self.rr[(topic, group)] += 1
29 self.inbox[m].append(msg)
30
31
32# ---------- 23.4 fan-out strategies ----------
33
34def make_world(n_users=20_000, n_authors=500, seed=23):
35 """Followers per author are very uneven: a few authors have most of them."""
36 rnd = random.Random(seed)
37 followers = defaultdict(set)
38 following = defaultdict(set)
39 weights = [1 / (i + 1) ** 1.2 for i in range(n_authors)]
40 for u in range(n_users):
41 for a in set(rnd.choices(range(n_authors), weights=weights, k=8)):
42 followers[a].add(u)
43 following[u].add(a)
44 return followers, following
45
46
47class FanoutOnWrite:
48 def __init__(self, followers):
49 self.followers = followers
50 self.feeds = defaultdict(list)
51 self.writes = 0
52
53 def post(self, author, book, t):
54 for u in self.followers[author]: # push to every follower's feed now
55 self.feeds[u].append((t, author, book))
56 self.writes += 1
57
58 def read(self, user, k=10):
59 return sorted(self.feeds[user], reverse=True)[:k]
60
61
62class FanoutOnRead:
63 def __init__(self, following):
64 self.following = following
65 self.posts = defaultdict(list)
66 self.reads = 0
67
68 def post(self, author, book, t):
69 self.posts[author].append((t, author, book))
70
71 def read(self, user, k=10):
72 items = []
73 for a in self.following[user]: # pull from everyone you follow, now
74 self.reads += 1
75 items.extend(self.posts[a])
76 return sorted(items, reverse=True)[:k]
77
78
79class Hybrid:
80 """Push for normal authors, pull for celebrities at read time."""
81
82 def __init__(self, followers, following, celebrity_threshold):
83 self.followers, self.following = followers, following
84 self.celebs = {a for a, f in followers.items() if len(f) >= celebrity_threshold}
85 self.feeds = defaultdict(list)
86 self.posts = defaultdict(list)
87 self.writes = self.reads = 0
88
89 def post(self, author, book, t):
90 self.posts[author].append((t, author, book))
91 if author in self.celebs:
92 return # don't fan out: too many followers
93 for u in self.followers[author]:
94 self.feeds[u].append((t, author, book))
95 self.writes += 1
96
97 def read(self, user, k=10):
98 items = list(self.feeds[user])
99 for a in self.following[user] & self.celebs:
100 self.reads += 1
101 items.extend(self.posts[a])
102 return sorted(items, reverse=True)[:k]
103
104
105# ---------- 23.6 duplicates and ordering ----------
106
107class StockView:
108 """Consumer that tolerates duplicates (event ids) and disorder (versions)."""
109
110 def __init__(self):
111 self.seen, self.stock, self.version = set(), {}, {}
112
113 def apply(self, event):
114 eid, book, version, qty = event
115 if eid in self.seen:
116 return "duplicate"
117 self.seen.add(eid)
118 if version <= self.version.get(book, 0):
119 return "stale"
120 self.version[book], self.stock[book] = version, qty
121 return "applied"
122
123
124# ---------- 23.7 outbox ----------
125
126class Db:
127 def __init__(self):
128 self.orders, self.outbox = {}, []
129
130 def place_order_with_outbox(self, order_id, fail=False):
131 staged_orders = dict(self.orders)
132 staged_outbox = list(self.outbox)
133 staged_orders[order_id] = "placed"
134 staged_outbox.append({"id": f"evt-{order_id}", "type": "order_placed",
135 "order_id": order_id, "sent": False})
136 if fail:
137 raise RuntimeError("rollback") # nothing above is kept
138 self.orders, self.outbox = staged_orders, staged_outbox # one commit
139
140
141def relay(db, broker_log, crash_before_mark=False):
142 """Publish unsent outbox rows, then mark them sent. A crash in between republishes."""
143 for row in db.outbox:
144 if not row["sent"]:
145 broker_log.append(row["id"])
146 if crash_before_mark:
147 return
148 row["sent"] = True
149
150
151def dual_write_publish_first(db_orders, broker_log, order_id, commit_fails):
152 broker_log.append(f"evt-{order_id}") # bug: published before the commit
153 if commit_fails:
154 return # order never saved
155 db_orders[order_id] = "placed"
156
157
158def dual_write_commit_first(db_orders, broker_log, order_id, crash_after_commit):
159 db_orders[order_id] = "placed"
160 if crash_after_commit:
161 return # event never published
162 broker_log.append(f"evt-{order_id}")
163
164
165# ---------- checks ----------
166
167def run_checks():
168 # pub/sub vs queue: three groups each get every event; two SMS workers share theirs
169 b = Broker()
170 b.subscribe("order_placed", "sms", "sms-1")
171 b.subscribe("order_placed", "sms", "sms-2")
172 b.subscribe("order_placed", "loyalty", "loyalty-1")
173 b.subscribe("order_placed", "analytics", "analytics-1")
174 for i in range(10):
175 b.publish("order_placed", i)
176 assert len(b.inbox["loyalty-1"]) == 10 and len(b.inbox["analytics-1"]) == 10
177 assert len(b.inbox["sms-1"]) == 5 and len(b.inbox["sms-2"]) == 5
178 assert sorted(b.inbox["sms-1"] + b.inbox["sms-2"]) == list(range(10))
179
180 # fan-out strategies on a skewed world
181 followers, following = make_world()
182 sizes = sorted((len(f) for f in followers.values()), reverse=True)
183 top = max(followers, key=lambda a: len(followers[a]))
184 assert sizes[0] > 50 * sizes[len(sizes) // 2] # very uneven
185
186 rnd = random.Random(7)
187 posts = [(rnd.choice(list(followers)), f"book{i}", i) for i in range(2_000)]
188 posts.append((top, "celebrity-new-book", 5_000))
189 w, r = FanoutOnWrite(followers), FanoutOnRead(following)
190 h = Hybrid(followers, following, celebrity_threshold=5_000)
191 for p in posts:
192 w.post(*p)
193 r.post(*p)
194 h.post(*p)
195 users = list(range(0, 20_000, 50)) # 400 feed loads
196 for u in users:
197 assert w.read(u) == r.read(u) == h.read(u) # same feeds, different cost
198 assert h.writes < 0.6 * w.writes # celebrities skipped: 44% fewer writes
199 assert h.reads < 0.4 * r.reads # only celebrities pulled: 66% fewer reads
200 fanned = len(followers[top])
201
202 # duplicates and disorder
203 view = StockView()
204 events = [("e1", "dune", 1, 10), ("e3", "dune", 3, 8), ("e2", "dune", 2, 9), ("e3", "dune", 3, 8)]
205 assert [view.apply(e) for e in events] == ["applied", "applied", "stale", "duplicate"]
206 assert view.stock["dune"] == 8
207
208 # outbox: all or nothing, and the relay publishes at least once
209 db, log = Db(), []
210 db.place_order_with_outbox("881")
211 try:
212 db.place_order_with_outbox("882", fail=True)
213 except RuntimeError:
214 pass
215 assert db.orders == {"881": "placed"} and len(db.outbox) == 1
216 relay(db, log, crash_before_mark=True) # published, not marked
217 relay(db, log) # restarted: publishes again
218 assert log == ["evt-881", "evt-881"] # at-least-once: consumers dedupe
219
220 # dual writes go wrong both ways
221 orders, out = {}, []
222 dual_write_publish_first(orders, out, "883", commit_fails=True)
223 assert out == ["evt-883"] and "883" not in orders # ghost event
224 orders, out = {}, []
225 dual_write_commit_first(orders, out, "884", crash_after_commit=True)
226 assert orders == {"884": "placed"} and out == [] # lost event
227
228 # Level 2 arithmetic
229 assert 5_000_000 / 50_000 == 100 # seconds to fan out a celebrity post
230 assert 300 * 1 == 300 # pulls per feed load (fan-out on read)
231
232 return {"top_followers": fanned, "writes": (w.writes, h.writes), "reads": (r.reads, h.reads),
233 "celebs": len(h.celebs)}
234
235
236if __name__ == "__main__":
237 print("All checks pass.", run_checks())