Publish-Subscribe Messaging

<h2 id="publish-subscribe-messaging" class="position-relative d-flex align-items-center group"> Publish-Subscribe Messaging <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="publish-subscribe-messaging" aria-haspopup="dialog" aria-label="Share link: Publish-Subscribe Messaging"> Share link </button> </h2><div id="headingShareModal" class="heading-share-modal" role="dialog" aria-modal="true" aria-labelledby="headingShareTitle" hidden> <div class="hsm-dialog" role="document"> <div class="hsm-header"> <h2 id="headingShareTitle" class="h6 mb-0 fw-bold">Share this section</h2> <button type="button" class="hsm-close" aria-label="Close"> </button> </div> <div class="hsm-body"> <label for="headingShareInput" class="form-label small text-muted mb-1 text-uppercase fw-bold" style="font-size: 0.7rem; letter-spacing: 0.5px;">Permalink</label> <div class="input-group mb-4 hsm-url-group"> <input id="headingShareInput" type="text" class="form-control font-monospace" readonly aria-readonly="true" style="font-size: 0.85rem;" /> <button class="btn btn-primary hsm-copy" type="button" aria-label="Copy" title="Copy"> </button> </div> <div class="small fw-bold mb-2 text-muted text-uppercase" style="font-size: 0.7rem; letter-spacing: 0.5px;">Share via</div> <div class="hsm-share-grid"> <a id="share-twitter" class="btn btn-outline-secondary w-100" target="_blank" rel="noopener noreferrer"> Twitter </a> <a id="share-linkedin" class="btn btn-outline-secondary w-100" target="_blank" rel="noopener noreferrer"> LinkedIn </a> <a id="share-facebook" class="btn btn-outline-secondary w-100" target="_blank" rel="noopener noreferrer"> Facebook </a> </div> </div> </div> </div> <style> .heading-share-modal { position: fixed; inset: 0; display: flex; justify-content: center; align-items: center; background: rgba(0, 0, 0, 0.6); z-index: 1050; padding: 1rem; backdrop-filter: blur(4px); -webkit-backdrop-filter: blur(4px); } .heading-share-modal[hidden] { display: none !important; } .hsm-dialog { max-width: 420px; width: 100%; background: var(--bs-body-bg, #fff); color: var(--bs-body-color, #212529); border: 1px solid var(--bs-border-color, rgba(0,0,0,0.1)); border-radius: 1rem; box-shadow: 0 25px 50px -12px rgba(0, 0, 0, 0.25); overflow: hidden; animation: hsm-fade-in 0.2s ease-out; } @keyframes hsm-fade-in { from { opacity: 0; transform: scale(0.95); } to { opacity: 1; transform: scale(1); } } [data-bs-theme="dark"] .hsm-dialog { background: #1e293b; border-color: rgba(255,255,255,0.1); color: #f8f9fa; } .hsm-header { display: flex; justify-content: space-between; align-items: center; padding: 1rem 1.5rem; border-bottom: 1px solid var(--bs-border-color, rgba(0,0,0,0.1)); background: rgba(0,0,0,0.02); } [data-bs-theme="dark"] .hsm-header { background: rgba(255,255,255,0.02); border-color: rgba(255,255,255,0.1); } .hsm-close { background: transparent; border: none; color: inherit; opacity: 0.5; padding: 0.25rem 0.5rem; border-radius: 0.25rem; font-size: 1.2rem; line-height: 1; transition: opacity 0.2s; } .hsm-close:hover { opacity: 1; } .hsm-body { padding: 1.5rem; } .hsm-url-group { display: flex !important; align-items: stretch; } .hsm-url-group .form-control { flex: 1; min-width: 0; margin: 0; background: var(--bs-secondary-bg, #f8f9fa); border-color: var(--bs-border-color, #dee2e6); border-top-right-radius: 0; border-bottom-right-radius: 0; height: 42px; } .hsm-url-group .btn { flex: 0 0 auto; margin: 0; margin-left: -1px; border-top-left-radius: 0; border-bottom-left-radius: 0; height: 42px; display: flex; align-items: center; justify-content: center; padding: 0 1.25rem; z-index: 2; } [data-bs-theme="dark"] .hsm-url-group .form-control { background: #0f172a; border-color: #334155; color: #e2e8f0; } .hsm-share-grid { display: flex; flex-direction: column; gap: 0.5rem; } .hsm-share-grid .btn { display: flex; align-items: center; justify-content: center; font-size: 0.9rem; padding: 0.6rem; border-color: var(--bs-border-color); width: 100%; } [data-bs-theme="dark"] .hsm-share-grid .btn { color: #e2e8f0; border-color: #475569; } [data-bs-theme="dark"] .hsm-share-grid .btn:hover { background: #334155; border-color: #cbd5e1; } </style> <script> (function(){ const modal = document.getElementById('headingShareModal'); if(!modal) return; const input = modal.querySelector('#headingShareInput'); const copyBtn = modal.querySelector('.hsm-copy'); const twitter = modal.querySelector('#share-twitter'); const linkedin = modal.querySelector('#share-linkedin'); const facebook = modal.querySelector('#share-facebook'); const closeBtn = modal.querySelector('.hsm-close'); let lastFocus=null; let trapBound=false; function buildUrl(id){ return window.location.origin + window.location.pathname + '#' + id; } function isOpen(){ return !modal.hasAttribute('hidden'); } function hydrate(id){ const url=buildUrl(id); input.value=url; const enc=encodeURIComponent(url); const text=encodeURIComponent(document.title); if(twitter) twitter.href=`https://twitter.com/intent/tweet?url=${enc}&text=${text}`; if(linkedin) linkedin.href=`https://www.linkedin.com/sharing/share-offsite/?url=${enc}`; if(facebook) facebook.href=`https://www.facebook.com/sharer/sharer.php?u=${enc}`; } function openModal(id){ lastFocus=document.activeElement; hydrate(id); if(!isOpen()){ modal.removeAttribute('hidden'); } requestAnimationFrame(()=>{ input.focus(); }); trapFocus(); } function closeModal(){ if(!isOpen()) return; modal.setAttribute('hidden',''); if(lastFocus && typeof lastFocus.focus==='function') lastFocus.focus(); } function copyCurrent(){ try{ navigator.clipboard.writeText(input.value).then(()=>feedback(true),()=>fallback()); } catch(e){ fallback(); } } function fallback(){ input.select(); try{ document.execCommand('copy'); feedback(true);}catch(e){ feedback(false);} } function feedback(ok){ if(!copyBtn) return; const icon=copyBtn.querySelector('i'); if(!icon) return; const prev=copyBtn.getAttribute('data-prev')||icon.className; if(!copyBtn.getAttribute('data-prev')) copyBtn.setAttribute('data-prev',prev); icon.className= ok ? 'fa-duotone fa-clipboard-check':'fa-duotone fa-circle-exclamation'; setTimeout(()=>{ icon.className=prev; },1800); } function handleShareClick(e){ e.preventDefault(); const btn=e.currentTarget; const id=btn.getAttribute('data-share-target'); if(id) openModal(id); } function bindShareButtons(){ document.querySelectorAll('.h-share').forEach(btn=>{ if(!btn.dataset.hShareBound){ btn.addEventListener('click', handleShareClick); btn.dataset.hShareBound='1'; } }); } bindShareButtons(); if(document.readyState==='loading'){ document.addEventListener('DOMContentLoaded', bindShareButtons); } else { requestAnimationFrame(bindShareButtons); } document.addEventListener('click', function(e){ const shareBtn=e.target.closest && e.target.closest('.h-share'); if(shareBtn && !shareBtn.dataset.hShareBound){ handleShareClick.call(shareBtn, e); } }, true); document.addEventListener('click', e=>{ if(e.target===modal) closeModal(); if(e.target.closest && e.target.closest('.hsm-close')){ e.preventDefault(); closeModal(); } if(copyBtn && (e.target===copyBtn || (e.target.closest && e.target.closest('.hsm-copy')))) { e.preventDefault(); copyCurrent(); } }); document.addEventListener('keydown', e=>{ if(e.key==='Escape' && isOpen()) closeModal(); }); function trapFocus(){ if(trapBound) return; trapBound=true; modal.addEventListener('keydown', f=>{ if(f.key==='Tab' && isOpen()){ const focusable=[...modal.querySelectorAll('a[href],button,input,textarea,select,[tabindex]:not([tabindex="-1"])')].filter(el=>!el.hasAttribute('disabled')); if(!focusable.length) return; const first=focusable[0]; const last=focusable[focusable.length-1]; if(f.shiftKey && document.activeElement===first){ f.preventDefault(); last.focus(); } else if(!f.shiftKey && document.activeElement===last){ f.preventDefault(); first.focus(); } } }); } if(closeBtn) closeBtn.addEventListener('click', e=>{ e.preventDefault(); closeModal(); }); })(); </script>Publish-Subscribe (Pub/Sub) is a messaging pattern where publishers send messages to topics without knowing who will receive them, and subscribers receive messages from topics they’re interested in without knowing who sent them. This decoupling enables scalable, flexible, event-driven architectures. <h3 id="pubsub-fundamentals" class="position-relative d-flex align-items-center group"> Pub/Sub Fundamentals <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="pubsub-fundamentals" aria-haspopup="dialog" aria-label="Share link: Pub/Sub Fundamentals"> Share link </button> </h3> <h4 id="core-concepts" class="position-relative d-flex align-items-center group"> Core Concepts <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="core-concepts" aria-haspopup="dialog" aria-label="Share link: Core Concepts"> Share link </button> </h4>Publishers - Send messages to topics Subscribers - Receive messages from topics Topics - Named channels for message distribution Messages - Data packets being exchanged <h4 id="benefits" class="position-relative d-flex align-items-center group"> Benefits <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="benefits" aria-haspopup="dialog" aria-label="Share link: Benefits"> Share link </button> </h4><ul> <li>Decoupling - Publishers and subscribers are independent</li> <li>Scalability - Add subscribers without affecting publishers</li> <li>Flexibility - Dynamic subscription management</li> <li>Fan-Out - One message reaches multiple subscribers</li> </ul> <h3 id="geode-pubsub" class="position-relative d-flex align-items-center group"> Geode Pub/Sub <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="geode-pubsub" aria-haspopup="dialog" aria-label="Share link: Geode Pub/Sub"> Share link </button> </h3> <h4 id="creating-topics" class="position-relative d-flex align-items-center group"> Creating Topics <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="creating-topics" aria-haspopup="dialog" aria-label="Share link: Creating Topics"> Share link </button> </h4>Define topics for message distribution: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Create topic for user events CREATE TOPIC user_events WITH OPTIONS ( retention_period: '7 days', message_ttl: '24 hours', max_subscribers: 100 ); -- Create topic with message schema CREATE TOPIC order_notifications SCHEMA ( order_id STRING, status STRING, timestamp TIMESTAMP, details MAP<STRING, ANY> ); </code></pre></div> <h4 id="publishing-messages" class="position-relative d-flex align-items-center group"> Publishing Messages <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="publishing-messages" aria-haspopup="dialog" aria-label="Share link: Publishing Messages"> Share link </button> </h4>Publish to topics: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">from geode_client import Client async def publish_user_event(client, user_id, event_type): """Publish user event to topic""" message = { "user_id": user_id, "event_type": event_type, "timestamp": datetime.now().isoformat(), "metadata": { "source": "user-service", "version": "1.0" } } await client.publish("user_events", message) print(f"Published: {event_type} for user {user_id}") </code></pre></div> <h4 id="subscribing-to-topics" class="position-relative d-flex align-items-center group"> Subscribing to Topics <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="subscribing-to-topics" aria-haspopup="dialog" aria-label="Share link: Subscribing to Topics"> Share link </button> </h4>Subscribe to receive messages: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def subscribe_to_user_events(client): """Subscribe to user event topic""" async with client.subscribe("user_events") as subscription: async for message in subscription: await handle_user_event(message) async def handle_user_event(message): """Process user event""" event_type = message["event_type"] user_id = message["user_id"] if event_type == "signup": await send_welcome_email(user_id) elif event_type == "purchase": await update_analytics(user_id, message) elif event_type == "churn": await trigger_retention_campaign(user_id) </code></pre></div> <h3 id="messaging-patterns" class="position-relative d-flex align-items-center group"> Messaging Patterns <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="messaging-patterns" aria-haspopup="dialog" aria-label="Share link: Messaging Patterns"> Share link </button> </h3> <h4 id="fan-out-pattern" class="position-relative d-flex align-items-center group"> Fan-Out Pattern <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="fan-out-pattern" aria-haspopup="dialog" aria-label="Share link: Fan-Out Pattern"> Share link </button> </h4>One message to many subscribers: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Publisher async def notify_all(client, notification): """Send notification to all subscribers""" await client.publish("notifications", { "title": notification.title, "body": notification.body, "priority": "high" }) # Multiple subscribers receive the same message async def email_subscriber(client): async with client.subscribe("notifications") as sub: async for msg in sub: await send_email(msg) async def sms_subscriber(client): async with client.subscribe("notifications") as sub: async for msg in sub: await send_sms(msg) async def push_subscriber(client): async with client.subscribe("notifications") as sub: async for msg in sub: await send_push_notification(msg) </code></pre></div> <h4 id="topic-filtering" class="position-relative d-flex align-items-center group"> Topic Filtering <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="topic-filtering" aria-haspopup="dialog" aria-label="Share link: Topic Filtering"> Share link </button> </h4>Subscribe with filters: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Subscribe only to high-priority orders SUBSCRIBE TO order_notifications WHERE priority = 'high'; -- Subscribe to specific user's events SUBSCRIBE TO user_events WHERE user_id = 'user123'; </code></pre></div><div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def filtered_subscription(client): """Subscribe with filter""" async with client.subscribe( "order_notifications", filter="priority = 'high'" ) as subscription: async for message in subscription: await handle_urgent_order(message) </code></pre></div> <h4 id="request-reply-pattern" class="position-relative d-flex align-items-center group"> Request-Reply Pattern <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="request-reply-pattern" aria-haspopup="dialog" aria-label="Share link: Request-Reply Pattern"> Share link </button> </h4>Implement RPC over pub/sub: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">import uuid class RPCClient: def __init__(self, client): self.client = client self.pending_requests = {} self.reply_topic = f"replies_{uuid.uuid4()}" async def call(self, method, params, timeout=5.0): """Make RPC call""" request_id = str(uuid.uuid4()) # Set up reply handler reply_future = asyncio.Future() self.pending_requests[request_id] = reply_future # Publish request await self.client.publish("rpc_requests", { "id": request_id, "method": method, "params": params, "reply_to": self.reply_topic }) # Wait for reply try: result = await asyncio.wait_for(reply_future, timeout) return result finally: del self.pending_requests[request_id] async def listen_for_replies(self): """Listen for RPC replies""" async with self.client.subscribe(self.reply_topic) as sub: async for message in sub: request_id = message["request_id"] if request_id in self.pending_requests: self.pending_requests[request_id].set_result(message["result"]) </code></pre></div> <h4 id="topic-hierarchies" class="position-relative d-flex align-items-center group"> Topic Hierarchies <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="topic-hierarchies" aria-haspopup="dialog" aria-label="Share link: Topic Hierarchies"> Share link </button> </h4>Organize topics hierarchically: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-fallback" data-lang="fallback">users/ users/created users/updated users/deleted users/verified orders/ orders/placed orders/shipped orders/delivered orders/canceled </code></pre></div><div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Subscribe to all user events async with client.subscribe("users/*") as sub: async for message in sub: handle_user_event(message) # Subscribe to specific event type async with client.subscribe("users/created") as sub: async for message in sub: handle_new_user(message) </code></pre></div> <h3 id="message-delivery-guarantees" class="position-relative d-flex align-items-center group"> Message Delivery Guarantees <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="message-delivery-guarantees" aria-haspopup="dialog" aria-label="Share link: Message Delivery Guarantees"> Share link </button> </h3> <h4 id="at-most-once" class="position-relative d-flex align-items-center group"> At-Most-Once <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="at-most-once" aria-haspopup="dialog" aria-label="Share link: At-Most-Once"> Share link </button> </h4>Message delivered zero or one time: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">CREATE TOPIC notifications WITH DELIVERY at_most_once; </code></pre></div><ul> <li>Fastest performance</li> <li>No retries</li> <li>Suitable for non-critical notifications</li> </ul> <h4 id="at-least-once" class="position-relative d-flex align-items-center group"> At-Least-Once <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="at-least-once" aria-haspopup="dialog" aria-label="Share link: At-Least-Once"> Share link </button> </h4>Message delivered one or more times: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">CREATE TOPIC orders WITH DELIVERY at_least_once; </code></pre></div><ul> <li>Messages may be duplicated</li> <li>Subscribers must be idempotent</li> <li>Suitable for most applications</li> </ul> <h4 id="exactly-once" class="position-relative d-flex align-items-center group"> Exactly-Once <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="exactly-once" aria-haspopup="dialog" aria-label="Share link: Exactly-Once"> Share link </button> </h4>Message delivered exactly one time: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">CREATE TOPIC transactions WITH DELIVERY exactly_once; </code></pre></div><ul> <li>Highest latency</li> <li>No duplicates</li> <li>Suitable for financial transactions</li> </ul> <h3 id="advanced-features" class="position-relative d-flex align-items-center group"> Advanced Features <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="advanced-features" aria-haspopup="dialog" aria-label="Share link: Advanced Features"> Share link </button> </h3> <h4 id="message-priorities" class="position-relative d-flex align-items-center group"> Message Priorities <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="message-priorities" aria-haspopup="dialog" aria-label="Share link: Message Priorities"> Share link </button> </h4>Prioritize message delivery: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Publish with priority await client.publish("tasks", { "task_id": "task123", "action": "process_payment" }, priority="high") # Subscribe to high-priority messages first async with client.subscribe( "tasks", priority_order=True ) as sub: async for message in sub: await process_task(message) </code></pre></div> <h4 id="message-expiration" class="position-relative d-flex align-items-center group"> Message Expiration <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="message-expiration" aria-haspopup="dialog" aria-label="Share link: Message Expiration"> Share link </button> </h4>Set message TTL: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Publish with TTL await client.publish("flash_sales", { "sale_id": "sale123", "discount": 0.5 }, ttl_seconds=3600) # Expire after 1 hour </code></pre></div> <h4 id="dead-letter-queues" class="position-relative d-flex align-items-center group"> Dead Letter Queues <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="dead-letter-queues" aria-haspopup="dialog" aria-label="Share link: Dead Letter Queues"> Share link </button> </h4>Handle failed messages: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">CREATE TOPIC payment_processing WITH OPTIONS ( max_retries: 3, dead_letter_topic: 'payment_failures' ); </code></pre></div><div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def process_failed_payments(client): """Handle messages that exceeded retry limit""" async with client.subscribe("payment_failures") as sub: async for failed_message in sub: await manual_review(failed_message) await notify_support(failed_message) </code></pre></div> <h4 id="message-batching" class="position-relative d-flex align-items-center group"> Message Batching <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="message-batching" aria-haspopup="dialog" aria-label="Share link: Message Batching"> Share link </button> </h4>Improve throughput with batching: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def batch_publisher(client): """Publish messages in batches""" messages = [ {"user_id": f"user{i}", "action": "login"} for i in range(1000) ] # Batch publish await client.publish_batch("user_events", messages) </code></pre></div> <h3 id="monitoring-and-observability" class="position-relative d-flex align-items-center group"> Monitoring and Observability <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="monitoring-and-observability" aria-haspopup="dialog" aria-label="Share link: Monitoring and Observability"> Share link </button> </h3> <h4 id="topic-metrics" class="position-relative d-flex align-items-center group"> Topic Metrics <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="topic-metrics" aria-haspopup="dialog" aria-label="Share link: Topic Metrics"> Share link </button> </h4>Monitor topic health: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">SELECT topic_name, messages_published, messages_consumed, pending_messages, subscriber_count, publish_rate, consume_rate FROM SYSTEM.topics WHERE pending_messages > 10000; </code></pre></div> <h4 id="subscriber-lag" class="position-relative d-flex align-items-center group"> Subscriber Lag <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="subscriber-lag" aria-haspopup="dialog" aria-label="Share link: Subscriber Lag"> Share link </button> </h4>Track consumer lag: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">SELECT subscriber_id, topic_name, lag_messages, lag_seconds, last_message_time FROM SYSTEM.subscribers WHERE lag_messages > 1000; </code></pre></div> <h3 id="best-practices" class="position-relative d-flex align-items-center group"> Best Practices <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="best-practices" aria-haspopup="dialog" aria-label="Share link: Best Practices"> Share link </button> </h3> <h4 id="message-design" class="position-relative d-flex align-items-center group"> Message Design <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="message-design" aria-haspopup="dialog" aria-label="Share link: Message Design"> Share link </button> </h4><ul> <li>Keep messages small and focused</li> <li>Include timestamp and metadata</li> <li>Version message schemas</li> <li>Make messages self-contained</li> <li>Use consistent naming conventions</li> </ul> <h4 id="subscription-management" class="position-relative d-flex align-items-center group"> Subscription Management <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="subscription-management" aria-haspopup="dialog" aria-label="Share link: Subscription Management"> Share link </button> </h4><ul> <li>Clean up unused subscriptions</li> <li>Monitor subscriber lag</li> <li>Implement backpressure handling</li> <li>Use appropriate delivery guarantees</li> <li>Handle redelivery idempotently</li> </ul> <h4 id="error-handling" class="position-relative d-flex align-items-center group"> Error Handling <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="error-handling" aria-haspopup="dialog" aria-label="Share link: Error Handling"> Share link </button> </h4><div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def robust_subscriber(client): """Handle errors gracefully""" while True: try: async with client.subscribe("events") as sub: async for message in sub: try: await process_message(message) except Exception as e: logger.error(f"Error processing message: {e}") await send_to_dlq(message, str(e)) except ConnectionError: await asyncio.sleep(5) # Reconnect delay </code></pre></div> <h3 id="related-topics" class="position-relative d-flex align-items-center group"> Related Topics <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="related-topics" aria-haspopup="dialog" aria-label="Share link: Related Topics"> Share link </button> </h3><ul> <li><a href="/tags/events" >Events</a> - Event-driven architecture</li> <li><a href="/tags/streaming" >Streaming</a> - Event streaming</li> <li><a href="/tags/cdc" >CDC</a> - Change data capture</li> <li><a href="/tags/websockets" >WebSockets</a> - Real-time communication</li> </ul> <h3 id="production-patterns" class="position-relative d-flex align-items-center group"> Production Patterns <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="production-patterns" aria-haspopup="dialog" aria-label="Share link: Production Patterns"> Share link </button> </h3> <h4 id="multi-region-pubsub" class="position-relative d-flex align-items-center group"> Multi-Region Pub/Sub <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="multi-region-pubsub" aria-haspopup="dialog" aria-label="Share link: Multi-Region Pub/Sub"> Share link </button> </h4>Distribute messages across geographic regions: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-yaml" data-lang="yaml">topics: global_notifications: replicas: - region: us-east priority: primary - region: eu-west priority: secondary - region: ap-south priority: secondary replication: async message_routing: geo_nearest # Route to nearest region </code></pre></div><div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Publisher in US publishes to local topic async with client.publish("global_notifications", message, region="us-east"): pass # Subscribers in EU/AP receive via replication async with client.subscribe("global_notifications", region="eu-west") as sub: async for msg in sub: # Receives messages from US with replication lag process_message(msg) </code></pre></div> <h4 id="message-acknowledgment-strategies" class="position-relative d-flex align-items-center group"> Message Acknowledgment Strategies <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="message-acknowledgment-strategies" aria-haspopup="dialog" aria-label="Share link: Message Acknowledgment Strategies"> Share link </button> </h4>Control message delivery and acknowledgment: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def manual_acknowledgment(): """Fine-grained control over message acknowledgment""" async with client.subscribe("orders", auto_ack=False) as sub: async for message in sub: try: # Process message await process_order(message.data) # Explicitly acknowledge await message.ack() logger.info(f"Processed and acked: {message.id}") except ProcessingError as e: # Negative acknowledgment (requeue) await message.nack(requeue=True) logger.error(f"Failed to process {message.id}: {e}") except FatalError as e: # Dead letter (don't requeue) await message.nack(requeue=False) logger.critical(f"Fatal error on {message.id}: {e}") </code></pre></div> <h4 id="consumer-groups" class="position-relative d-flex align-items-center group"> Consumer Groups <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="consumer-groups" aria-haspopup="dialog" aria-label="Share link: Consumer Groups"> Share link </button> </h4>Load balance message processing across consumers: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Consumer group for parallel processing async def consumer_group_worker(worker_id, group_id): """Worker in consumer group""" async with client.subscribe( "order_processing", consumer_group=group_id, # Messages distributed across group consumer_id=worker_id ) as sub: async for message in sub: logger.info(f"Worker {worker_id} processing {message.id}") await process_order(message.data) await message.ack() # Start multiple workers (messages distributed among them) async def main(): workers = [ consumer_group_worker(f"worker-{i}", "order-processors") for i in range(10) # 10 workers share the load ] await asyncio.gather(*workers) </code></pre></div> <h4 id="ordered-message-processing" class="position-relative d-flex align-items-center group"> Ordered Message Processing <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="ordered-message-processing" aria-haspopup="dialog" aria-label="Share link: Ordered Message Processing"> Share link </button> </h4>Maintain message ordering when required: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Publish with partition key (same key → same partition → ordered) await client.publish("user_events", { "user_id": "user123", "action": "login", "timestamp": datetime.now() }, partition_key="user123") # All events for user123 ordered # Subscribe with ordering guarantee async with client.subscribe( "user_events", ordering=True, partition_key_filter="user123" # Process only this user's events in order ) as sub: async for event in sub: # Events for user123 arrive in published order await process_user_event(event) </code></pre></div> <h3 id="advanced-messaging-patterns" class="position-relative d-flex align-items-center group"> Advanced Messaging Patterns <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="advanced-messaging-patterns" aria-haspopup="dialog" aria-label="Share link: Advanced Messaging Patterns"> Share link </button> </h3> <h4 id="saga-pattern-with-pubsub" class="position-relative d-flex align-items-center group"> Saga Pattern with Pub/Sub <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="saga-pattern-with-pubsub" aria-haspopup="dialog" aria-label="Share link: Saga Pattern with Pub/Sub"> Share link </button> </h4>Coordinate distributed transactions: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class OrderSaga: """Distributed transaction using pub/sub""" def __init__(self, client): self.client = client async def start_order(self, order_data): """Initiate saga""" saga_id = str(uuid.uuid4()) # Publish order created event await self.client.publish("saga_events", { "saga_id": saga_id, "type": "ORDER_CREATED", "data": order_data }) return saga_id async def saga_orchestrator(self): """Orchestrate saga steps""" async with self.client.subscribe("saga_events") as sub: async for event in sub: saga_id = event.data['saga_id'] event_type = event.data['type'] if event_type == "ORDER_CREATED": # Step 1: Reserve inventory await self.reserve_inventory(saga_id, event.data['data']) elif event_type == "INVENTORY_RESERVED": # Step 2: Process payment await self.process_payment(saga_id, event.data['data']) elif event_type == "PAYMENT_PROCESSED": # Step 3: Ship order await self.ship_order(saga_id, event.data['data']) elif event_type == "SHIPPING_CONFIRMED": # Saga complete await self.complete_saga(saga_id) elif event_type.endswith("_FAILED"): # Compensating actions await self.rollback_saga(saga_id, event_type) async def reserve_inventory(self, saga_id, order): """Reserve inventory (Step 1)""" try: # Business logic result = await inventory_service.reserve(order['items']) # Publish success event await self.client.publish("saga_events", { "saga_id": saga_id, "type": "INVENTORY_RESERVED", "data": result }) except Exception as e: # Publish failure event await self.client.publish("saga_events", { "saga_id": saga_id, "type": "INVENTORY_RESERVATION_FAILED", "error": str(e) }) async def rollback_saga(self, saga_id, failed_step): """Compensating transactions""" if failed_step in ["PAYMENT_FAILED", "SHIPPING_FAILED"]: # Release inventory reservation await self.release_inventory(saga_id) if failed_step == "SHIPPING_FAILED": # Refund payment await self.refund_payment(saga_id) # Mark saga as rolled back await self.client.publish("saga_events", { "saga_id": saga_id, "type": "SAGA_ROLLED_BACK", "reason": failed_step }) </code></pre></div> <h4 id="event-sourcing" class="position-relative d-flex align-items-center group"> Event Sourcing <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-sourcing" aria-haspopup="dialog" aria-label="Share link: Event Sourcing"> Share link </button> </h4>Store state as sequence of events: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class EventSourcedEntity: """Entity rebuilt from event stream""" def __init__(self, entity_id, client): self.entity_id = entity_id self.client = client self.state = {} self.version = 0 async def load(self): """Rebuild state from events""" # Subscribe to entity's event stream async with self.client.subscribe( f"entity_{self.entity_id}", from_beginning=True # Replay all events ) as sub: async for event in sub: self.apply_event(event) self.version += 1 def apply_event(self, event): """Apply event to state""" if event['type'] == 'ACCOUNT_CREATED': self.state['balance'] = event['initial_balance'] self.state['created_at'] = event['timestamp'] elif event['type'] == 'DEPOSIT': self.state['balance'] += event['amount'] elif event['type'] == 'WITHDRAWAL': self.state['balance'] -= event['amount'] async def deposit(self, amount): """Command: deposit money""" # Validate if amount <= 0: raise ValueError("Amount must be positive") # Publish event await self.client.publish(f"entity_{self.entity_id}", { "type": "DEPOSIT", "amount": amount, "timestamp": datetime.now(), "version": self.version + 1 }) # Apply to local state self.state['balance'] += amount self.version += 1 </code></pre></div> <h4 id="cqrs-with-pubsub" class="position-relative d-flex align-items-center group"> CQRS with Pub/Sub <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="cqrs-with-pubsub" aria-haspopup="dialog" aria-label="Share link: CQRS with Pub/Sub"> Share link </button> </h4>Separate read and write models: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Write side: publish events async def create_user(user_data): """Command: create user (write model)""" user_id = str(uuid.uuid4()) # Store in write database async with geode_client.connection() as conn: await conn.begin() try: await conn.execute(""" CREATE (u:User {id: $id, email: $email, created_at: datetime()}) """, {"id": user_id, "email": user_data['email']}) await conn.commit() except Exception: await conn.rollback() raise # Publish event await pubsub_client.publish("user_events", { "type": "USER_CREATED", "user_id": user_id, "data": user_data }) return user_id # Read side: consume events and update read model async def user_projection_updater(): """Update read-optimized view""" async with pubsub_client.subscribe("user_events") as sub: async for event in sub: if event['type'] == 'USER_CREATED': # Update read database (denormalized for fast queries) await redis.hset( f"user:{event['user_id']}", mapping={ "email": event['data']['email'], "created_at": event['data']['created_at'] } ) elif event['type'] == 'USER_UPDATED': # Update read model await redis.hset( f"user:{event['user_id']}", mapping=event['data'] ) # Read from read model (fast) async def get_user(user_id): """Query: get user (read model)""" return await redis.hgetall(f"user:{user_id}") </code></pre></div> <h3 id="performance-tuning" class="position-relative d-flex align-items-center group"> Performance Tuning <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="performance-tuning" aria-haspopup="dialog" aria-label="Share link: Performance Tuning"> Share link </button> </h3> <h4 id="batch-publishing" class="position-relative d-flex align-items-center group"> Batch Publishing <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="batch-publishing" aria-haspopup="dialog" aria-label="Share link: Batch Publishing"> Share link </button> </h4>Improve throughput with batching: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def batch_publish_example(): """Publish messages in batches""" messages = [ {"user_id": i, "action": "click", "timestamp": datetime.now()} for i in range(10000) ] # Batch publish (100x faster than individual publishes) await client.publish_batch("user_actions", messages, batch_size=100) # Publishes 100 messages per network round-trip </code></pre></div> <h4 id="message-compression" class="position-relative d-flex align-items-center group"> Message Compression <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="message-compression" aria-haspopup="dialog" aria-label="Share link: Message Compression"> Share link </button> </h4>Reduce network bandwidth: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Enable compression for large messages await client.publish( "analytics_data", large_payload, compression="lz4" # Fast compression algorithm ) # Compression reduces bandwidth by 70-90% for JSON payloads </code></pre></div> <h4 id="topic-sharding" class="position-relative d-flex align-items-center group"> Topic Sharding <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="topic-sharding" aria-haspopup="dialog" aria-label="Share link: Topic Sharding"> Share link </button> </h4>Scale topics horizontally: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-yaml" data-lang="yaml"># Shard topic across multiple partitions topics: high_volume_events: partitions: 64 # 64 shards for parallel processing partition_key: user_id # Partition by user ID </code></pre></div><div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Messages automatically sharded by partition key await client.publish( "high_volume_events", message, partition_key=user_id # Determines which partition ) # Subscribers process partitions in parallel async def consume_partition(partition_id): async with client.subscribe( "high_volume_events", partition=partition_id ) as sub: async for msg in sub: await process_message(msg) # 64 workers, each processing one partition workers = [consume_partition(i) for i in range(64)] await asyncio.gather(*workers) </code></pre></div> <h3 id="monitoring-and-operations" class="position-relative d-flex align-items-center group"> Monitoring and Operations <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="monitoring-and-operations" aria-haspopup="dialog" aria-label="Share link: Monitoring and Operations"> Share link </button> </h3> <h4 id="message-flow-metrics" class="position-relative d-flex align-items-center group"> Message Flow Metrics <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="message-flow-metrics" aria-haspopup="dialog" aria-label="Share link: Message Flow Metrics"> Share link </button> </h4>Track message flow through system: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">from prometheus_client import Counter, Histogram messages_published = Counter( 'pubsub_messages_published_total', 'Total messages published', ['topic'] ) messages_consumed = Counter( 'pubsub_messages_consumed_total', 'Total messages consumed', ['topic', 'consumer_group'] ) message_latency = Histogram( 'pubsub_message_latency_seconds', 'Message end-to-end latency', ['topic'] ) async def publish_with_metrics(topic, message): """Publish with metrics tracking""" start_time = time.time() message['_published_at'] = start_time await client.publish(topic, message) messages_published.labels(topic=topic).inc() async def consume_with_metrics(topic, consumer_group): """Consume with metrics tracking""" async with client.subscribe(topic, consumer_group=consumer_group) as sub: async for msg in sub: # Track latency latency = time.time() - msg.data['_published_at'] message_latency.labels(topic=topic).observe(latency) # Process await process_message(msg) messages_consumed.labels( topic=topic, consumer_group=consumer_group ).inc() </code></pre></div> <h4 id="health-checks" class="position-relative d-flex align-items-center group"> Health Checks <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="health-checks" aria-haspopup="dialog" aria-label="Share link: Health Checks"> Share link </button> </h4>Monitor pub/sub system health: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def health_check(): """Check pub/sub system health""" # Check topic lag lag = await client.get_consumer_lag("critical_events", "processors") if lag > 10000: alert("High consumer lag: {}".format(lag)) # Check error rates error_rate = await client.get_error_rate("critical_events") if error_rate > 0.01: # >1% error rate alert("High error rate: {:.2%}".format(error_rate)) # Check throughput throughput = await client.get_throughput("critical_events") if throughput < 100: # <100 msgs/sec alert("Low throughput: {} msgs/sec".format(throughput)) </code></pre></div> <h3 id="learn-more" class="position-relative d-flex align-items-center group"> Learn More <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="learn-more" aria-haspopup="dialog" aria-label="Share link: Learn More"> Share link </button> </h3><ul> <li><a href="https://docs.microsoft.com/en-us/azure/architecture/patterns/publisher-subscriber" aria-label="Pub/Sub Patterns – opens in new window" target="_blank" rel="noopener noreferrer" >Pub/Sub Patterns ↗ </a> </li> <li><a href="https://cloud.google.com/pubsub/docs" aria-label="Google Cloud Pub/Sub – opens in new window" target="_blank" rel="noopener noreferrer" >Google Cloud Pub/Sub ↗ </a> </li> <li><a href="https://kafka.apache.org/" aria-label="Apache Kafka – opens in new window" target="_blank" rel="noopener noreferrer" >Apache Kafka ↗ </a> </li> <li><a href="https://martinfowler.com/articles/201701-event-driven.html" aria-label="Event-Driven Architecture – opens in new window" target="_blank" rel="noopener noreferrer" >Event-Driven Architecture ↗ </a> </li> <li><a href="https://microservices.io/patterns/data/saga.html" aria-label="Saga Pattern – opens in new window" target="_blank" rel="noopener noreferrer" >Saga Pattern ↗ </a> </li> <li><a href="https://martinfowler.com/eaaDev/EventSourcing.html" aria-label="Event Sourcing – opens in new window" target="_blank" rel="noopener noreferrer" >Event Sourcing ↗ </a> </li> </ul>

Popular

Related Articles