Architecture

When data changes in the source-of-truth service, a cache invalidation event is published. All services with local caches subscribe and evict stale entries in real time.

One invalidation event fans out to every cache listener; each evicts its own stale entries.

Implementation

Cache Invalidation Publisher

When a product is updated, publish an invalidation event with the affected cache keys.

cache_invalidation_publisher.go

package main

import (
    "context"
    "encoding/json"
    "log"

    "github.com/kubemq-io/kubemq-go/v2"
)

type CacheInvalidation struct {
    Entity string   `json:"entity"`
    Keys   []string `json:"keys"`
    Action string   `json:"action"`
}

func invalidateCache(ctx context.Context, client *kubemq.Client, inv CacheInvalidation) error {
    body, _ := json.Marshal(inv)
    return client.SendEvent(ctx, kubemq.NewEvent().
        SetChannel("cache.invalidate."+inv.Entity).
        SetMetadata("cache.invalidate").
        SetBody(body).
        SetTags(map[string]string{"entity": inv.Entity, "action": inv.Action}),
    )
}

func main() {
    ctx := context.Background()
    client, err := kubemq.NewClient(ctx,
        kubemq.WithAddress("localhost", 50000),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    err = invalidateCache(ctx, client, CacheInvalidation{
        Entity: "products",
        Keys:   []string{"product:SKU-100", "product:SKU-101"},
        Action: "update",
    })
    if err != nil {
        log.Fatal(err)
    }
    log.Println("Cache invalidation event published")
}

cache_invalidation_publisher.py

import json
from kubemq.pubsub import Client as PubSubClient
from kubemq.pubsub import EventMessage

client = PubSubClient(address="localhost:50000")

invalidation = {
    "entity": "products",
    "keys": ["product:SKU-100", "product:SKU-101"],
    "action": "update",
}

client.send_event(
    EventMessage(
        channel=f"cache.invalidate.{invalidation['entity']}",
        metadata="cache.invalidate",
        body=json.dumps(invalidation).encode("utf-8"),
        tags={"entity": invalidation["entity"], "action": invalidation["action"]},
    )
)
print("Cache invalidation event published")
client.close()

cache_invalidation_publisher.js

import { KubeMQClient, createEventMessage } from "kubemq-js";

const client = await KubeMQClient.create({
  address: "localhost:50000",
  clientId: "cache-invalidator",
});

const invalidation = {
  entity: "products",
  keys: ["product:SKU-100", "product:SKU-101"],
  action: "update",
};

await client.sendEvent(
  createEventMessage({
    channel: `cache.invalidate.${invalidation.entity}`,
    metadata: "cache.invalidate",
    body: JSON.stringify(invalidation),
    tags: { entity: invalidation.entity, action: invalidation.action },
  }),
);

console.log("Cache invalidation event published");
await client.close();

CacheInvalidationPublisher.java

PubSubClient client = PubSubClient.builder()
    .address("localhost:50000")
    .clientId("cache-invalidator")
    .build();

String body = "{\"entity\":\"products\","
    + "\"keys\":[\"product:SKU-100\",\"product:SKU-101\"],"
    + "\"action\":\"update\"}";

client.sendEventsMessage(EventMessage.builder()
    .channel("cache.invalidate.products")
    .metadata("cache.invalidate")
    .body(body.getBytes())
    .tags(Map.of("entity", "products", "action", "update"))
    .build());

System.out.println("Cache invalidation event published");
client.close();

CacheInvalidationPublisher.cs

await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

var invalidation = new
{
    entity = "products",
    keys = new[] { "product:SKU-100", "product:SKU-101" },
    action = "update"
};

await client.SendEventAsync(new EventMessage
{
    Channel = "cache.invalidate.products",
    Metadata = "cache.invalidate",
    Body = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(invalidation)),
    Tags = new Dictionary<string, string>
    {
        ["entity"] = "products", ["action"] = "update"
    }
});

Console.WriteLine("Cache invalidation event published");

CacheInvalidationPublisher.kt

val client = PubSubClient("localhost:50000")

val body = """{"entity":"products","keys":["product:SKU-100","product:SKU-101"],"action":"update"}"""

client.sendEvent(EventMessage(
    channel = "cache.invalidate.products",
    metadata = "cache.invalidate",
    body = body.toByteArray(),
    tags = mapOf("entity" to "products", "action" to "update"),
))

println("Cache invalidation event published")
client.close()

cache_invalidation_publisher.cpp

auto client = kubemq::PubSubClient("localhost:50000");

kubemq::EventMessage event;
event.channel = "cache.invalidate.products";
event.metadata = "cache.invalidate";
event.body = R"({"entity":"products","keys":["product:SKU-100","product:SKU-101"],"action":"update"})";
event.tags["entity"] = "products";
event.tags["action"] = "update";

client.sendEvent(event);
std::cout << "Cache invalidation event published" << std::endl;

cache_invalidation_publisher.rs

use kubemq::prelude::*;
use kubemq::EventBuilder;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .client_id("cache-invalidator")
        .build()
        .await?;

    let body = r#"{"entity":"products","keys":["product:SKU-100","product:SKU-101"],"action":"update"}"#;

    let event = EventBuilder::new()
        .channel("cache.invalidate.products")
        .metadata("cache.invalidate")
        .body(body.as_bytes().to_vec())
        .add_tag("entity", "products")
        .add_tag("action", "update")
        .build();

    client.send_event(event).await?;
    println!("Cache invalidation event published");

    client.close().await?;
    Ok(())
}

cache_invalidation_publisher.rb

require 'kubemq'

client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'cache-invalidator')

invalidation = '{"entity":"products","keys":["product:SKU-100","product:SKU-101"],"action":"update"}'

msg = KubeMQ::PubSub::EventMessage.new(
  channel: 'cache.invalidate.products',
  metadata: 'cache.invalidate',
  body: invalidation,
  tags: { 'entity' => 'products', 'action' => 'update' }
)

client.send_event(msg)
puts 'Cache invalidation event published'
client.close

cache_invalidation_publisher.exs

{:ok, client} =
  KubeMQ.Client.start_link(address: "localhost:50000", client_id: "cache-invalidator")

body = ~s({"entity":"products","keys":["product:SKU-100","product:SKU-101"],"action":"update"})

event =
  KubeMQ.Event.new(
    channel: "cache.invalidate.products",
    metadata: "cache.invalidate",
    body: body,
    tags: %{"entity" => "products", "action" => "update"}
  )

case KubeMQ.Client.send_event(client, event) do
  :ok -> IO.puts("Cache invalidation event published")
  {:error, err} -> IO.puts("Send failed: #{err.message}")
end

KubeMQ.Client.close(client)

Cache Listener

Each service subscribes to invalidation events and evicts matching entries from its local cache.

cache_listener.go

type LocalCache struct {
    mu    sync.RWMutex
    store map[string]interface{}
}

func (c *LocalCache) Evict(keys []string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    for _, key := range keys {
        delete(c.store, key)
        fmt.Printf("[Cache] Evicted: %s\n", key)
    }
}

cache := &LocalCache{store: make(map[string]interface{})}

sub, err := client.SubscribeToEvents(ctx, "cache.invalidate.>", "",
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        var inv CacheInvalidation
        json.Unmarshal(event.Body, &inv)
        cache.Evict(inv.Keys)
    }),
    kubemq.WithOnError(func(err error) {
        log.Println("[Cache] Error:", err)
    }),
)

cache_listener.py

import json

local_cache = {}

def on_invalidation(event):
    data = json.loads(event.body.decode("utf-8"))
    for key in data["keys"]:
        local_cache.pop(key, None)
        print(f"[Cache] Evicted: {key}")

client.subscribe_to_events(
    subscription=EventsSubscription(
        channel="cache.invalidate.>",
        on_receive_event_callback=on_invalidation,
        on_error_callback=lambda e: print(f"[Cache] Error: {e}"),
    ),
    cancel=CancellationToken(),
)

cache_listener.js

const localCache = new Map();

client.subscribeToEvents({
  channel: "cache.invalidate.>",
  onEvent: (msg) => {
    const data = JSON.parse(Buffer.from(msg.body).toString());
    for (const key of data.keys) {
      localCache.delete(key);
      console.log(`[Cache] Evicted: ${key}`);
    }
  },
  onError: (err) => console.error("[Cache] Error:", err.message),
});

CacheListener.java

ConcurrentHashMap<String, Object> localCache = new ConcurrentHashMap<>();

client.subscribeToEvents(EventsSubscription.builder()
    .channel("cache.invalidate.>")
    .onReceiveEventCallback(event -> {
        String body = new String(event.getBody());
        // Extract keys and evict
        List<String> keys = parseKeys(body);
        for (String key : keys) {
            localCache.remove(key);
            System.out.println("[Cache] Evicted: " + key);
        }
    })
    .onErrorCallback(err ->
        System.err.println("[Cache] Error: " + err.getMessage()))
    .build());

CacheListener.cs

var localCache = new ConcurrentDictionary<string, object>();

await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = "cache.invalidate.>" }))
{
    var data = JsonSerializer.Deserialize<JsonElement>(msg.Body.Span);
    foreach (var key in data.GetProperty("keys").EnumerateArray())
    {
        localCache.TryRemove(key.GetString()!, out _);
        Console.WriteLine($"[Cache] Evicted: {key.GetString()}");
    }
}

CacheListener.kt

val localCache = ConcurrentHashMap<String, Any>()

client.subscribeToEvents(
    channel = "cache.invalidate.>",
    onEvent = { event ->
        val body = String(event.body)
        val keysMatch = Regex(""""keys":\[(.*?)]""").find(body)
        keysMatch?.groupValues?.get(1)?.split(",")?.forEach { key ->
            val cleanKey = key.trim().removeSurrounding("\"")
            localCache.remove(cleanKey)
            println("[Cache] Evicted: $cleanKey")
        }
    },
    onError = { err -> System.err.println("[Cache] Error: ${err.message}") }
)

cache_listener.cpp

std::map<std::string, std::string> localCache;

client.subscribeToEvents("cache.invalidate.>", "",
    [&localCache](const kubemq::Event& event) {
        // Parse keys from JSON and evict
        // Simplified: evict all keys matching entity
        std::cout << "[Cache] Processing invalidation: "
                  << event.body << std::endl;
    },
    [](const std::string& err) {
        std::cerr << "[Cache] Error: " << err << std::endl;
    }
);

cache_listener.rs

use kubemq::prelude::*;
use serde_json::Value;

// Subscribe to every cache.invalidate.* channel and evict matching keys.
let sub = client
    .subscribe_to_events(
        "cache.invalidate.*",
        "",
        |event| {
            Box::pin(async move {
                let data: Value =
                    serde_json::from_slice(&event.body).unwrap_or(Value::Null);
                if let Some(keys) = data["keys"].as_array() {
                    for key in keys {
                        if let Some(k) = key.as_str() {
                            // local_cache.remove(k);
                            println!("[Cache] Evicted: {}", k);
                        }
                    }
                }
            })
        },
        None,
    )
    .await?;

cache_listener.rb

require 'json'

local_cache = {}
cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::PubSub::EventsSubscription.new(channel: 'cache.invalidate.*')
client.subscribe_to_events(sub, cancellation_token: cancel, on_error: ->(e) { puts "[Cache] Error: #{e.message}" }) do |event|
  data = JSON.parse(event.body)
  data['keys'].each do |key|
    local_cache.delete(key)
    puts "[Cache] Evicted: #{key}"
  end
end

cache_listener.exs

# local_cache is an Agent or ETS table holding the cached entries
{:ok, sub} =
  KubeMQ.Client.subscribe_to_events(client, "cache.invalidate.*",
    on_event: fn event ->
      %{"keys" => keys} = Jason.decode!(event.body)

      Enum.each(keys, fn key ->
        # Agent.update(local_cache, &Map.delete(&1, key))
        IO.puts("[Cache] Evicted: #{key}")
      end)
    end,
    on_error: fn err -> IO.puts("[Cache] Error: #{err.message}") end
  )

Production Considerations

Wildcard Subscriptions for selective cache listening
Multicast Events for broadcasting to events and queues simultaneously
Events Store for guaranteed cache invalidation delivery

Cache Invalidation

Architecture

Implementation

Cache Invalidation Publisher

Cache Listener

Production Considerations

What if a cache listener misses an event?

Thundering herd on cache miss

Selective invalidation

Related

On this page