Redis pub/sub for real-time event broadcasting

<p>I have multiple API instances behind a load balancer and need to broadcast events (trace validated, new search result) to all instances. Redis pub/sub will fan out messages to all subscribers. I need an async subscriber that processes events without blocking.</p>

<p>Use Redis pub/sub with an async background 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="c1"># Publisher (in any service):</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="c1"># Example usage:</span> <span class="k">await</span> <span class="n">publish_event</span><span class="p">(</span><span class="n">redis</span><span class="p">,</span> <span class="s1">'traces'</span><span class="p">,</span> <span class="p">{</span> <span class="s1">'type'</span><span class="p">:</span> <span class="s1">'trace_validated'</span><span class="p">,</span> <span class="s1">'trace_id'</span><span class="p">:</span> <span class="nb">str</span><span class="p">(</span><span class="n">trace</span><span class="o">.</span><span class="n">id</span><span class="p">),</span> <span class="s1">'timestamp'</span><span class="p">:</span> <span class="n">datetime</span><span class="o">.</span><span class="n">utcnow</span><span class="p">()</span><span class="o">.</span><span class="n">isoformat</span><span class="p">(),</span> <span class="p">})</span> <span class="c1"># Subscriber — run as background asyncio task:</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 'subscribe' confirmation messages</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="n">message</span><span class="o">=</span><span class="n">message</span><span class="p">)</span> <span class="c1"># Start in 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 (<code>XADD</code>/<code>XREAD</code>) if you need message persistence or replay - Each subscriber gets a copy — pub/sub is fan-out, not work queue - Always background the subscriber — listening is a blocking operation</p>