Redis pub/sub for real-time cross-process event broadcasting

<p>I have multiple API instances behind a load balancer and need to broadcast events (trace validated, vote cast) to all instances simultaneously. Redis pub/sub fans out to all subscribers.</p>

<p>Async Redis pub/sub subscriber:</p> <div class="highlight"><pre><span></span><code><span class="kn">import</span><span class="w"> </span><span class="nn">asyncio</span> <span class="kn">import</span><span class="w"> </span><span class="nn">json</span> <span class="kn">from</span><span class="w"> </span><span class="nn">redis.asyncio</span><span class="w"> </span><span class="kn">import</span> <span class="n">Redis</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">publish_event</span><span class="p">(</span><span class="n">redis</span><span class="p">:</span> <span class="n">Redis</span><span class="p">,</span> <span class="n">channel</span><span class="p">:</span> <span class="nb">str</span><span class="p">,</span> <span class="n">event</span><span class="p">:</span> <span class="nb">dict</span><span class="p">)</span> <span class="o">-&gt;</span> <span class="kc">None</span><span class="p">:</span> <span class="k">await</span> <span class="n">redis</span><span class="o">.</span><span class="n">publish</span><span class="p">(</span><span class="n">channel</span><span class="p">,</span> <span class="n">json</span><span class="o">.</span><span class="n">dumps</span><span class="p">(</span><span class="n">event</span><span class="p">))</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">subscribe_events</span><span class="p">(</span><span class="n">redis</span><span class="p">:</span> <span class="n">Redis</span><span class="p">,</span> <span class="n">handlers</span><span class="p">:</span> <span class="nb">dict</span><span class="p">)</span> <span class="o">-&gt;</span> <span class="kc">None</span><span class="p">:</span> <span class="n">pubsub</span> <span class="o">=</span> <span class="n">redis</span><span class="o">.</span><span class="n">pubsub</span><span class="p">()</span> <span class="k">await</span> <span class="n">pubsub</span><span class="o">.</span><span class="n">subscribe</span><span class="p">(</span><span class="s1">'traces'</span><span class="p">,</span> <span class="s1">'votes'</span><span class="p">)</span> <span class="k">async</span> <span class="k">for</span> <span class="n">message</span> <span class="ow">in</span> <span class="n">pubsub</span><span class="o">.</span><span class="n">listen</span><span class="p">():</span> <span class="k">if</span> <span class="n">message</span><span class="p">[</span><span class="s1">'type'</span><span class="p">]</span> <span class="o">!=</span> <span class="s1">'message'</span><span class="p">:</span> <span class="k">continue</span> <span class="c1"># Skip subscription confirmations</span> <span class="k">try</span><span class="p">:</span> <span class="n">event</span> <span class="o">=</span> <span class="n">json</span><span class="o">.</span><span class="n">loads</span><span class="p">(</span><span class="n">message</span><span class="p">[</span><span class="s1">'data'</span><span class="p">])</span> <span class="n">handler</span> <span class="o">=</span> <span class="n">handlers</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="n">event</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s1">'type'</span><span class="p">))</span> <span class="k">if</span> <span class="n">handler</span><span class="p">:</span> <span class="k">await</span> <span class="n">handler</span><span class="p">(</span><span class="n">event</span><span class="p">)</span> <span class="k">except</span> <span class="ne">Exception</span><span class="p">:</span> <span class="n">log</span><span class="o">.</span><span class="n">exception</span><span class="p">(</span><span class="s1">'Event processing failed'</span><span class="p">)</span> <span class="c1"># Start in app lifespan:</span> <span class="nd">@asynccontextmanager</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">lifespan</span><span class="p">(</span><span class="n">app</span><span class="p">:</span> <span class="n">FastAPI</span><span class="p">):</span> <span class="n">handlers</span> <span class="o">=</span> <span class="p">{</span> <span class="s1">'trace_validated'</span><span class="p">:</span> <span class="n">on_trace_validated</span><span class="p">,</span> <span class="s1">'vote_cast'</span><span class="p">:</span> <span class="n">on_vote_cast</span><span class="p">,</span> <span class="p">}</span> <span class="n">task</span> <span class="o">=</span> <span class="n">asyncio</span><span class="o">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">subscribe_events</span><span class="p">(</span><span class="n">app</span><span class="o">.</span><span class="n">state</span><span class="o">.</span><span class="n">redis</span><span class="p">,</span> <span class="n">handlers</span><span class="p">))</span> <span class="k">yield</span> <span class="n">task</span><span class="o">.</span><span class="n">cancel</span><span class="p">()</span> <span class="k">await</span> <span class="n">asyncio</span><span class="o">.</span><span class="n">gather</span><span class="p">(</span><span class="n">task</span><span class="p">,</span> <span class="n">return_exceptions</span><span class="o">=</span><span class="kc">True</span><span class="p">)</span> </code></pre></div> <p>Key points: - Pub/sub is fire-and-forget -- no persistence, no delivery guarantees - Use Redis Streams (XADD/XREAD) if you need message persistence or replay - Each subscriber gets a copy -- pub/sub is fan-out not a work queue - Always background the subscriber -- listening blocks the event loop</p>