# Cache Invalidation (/learn/events/scenarios/cache-invalidation)



## Architecture [#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.

<Mermaid
  chart="graph LR
  PS[&#x22;Product Service<br/>(source of truth)&#x22;]
  EV{{&#x22;Events channel<br/>cache.invalidate.*&#x22;}}
  API1[&#x22;API Gateway Cache&#x22;]
  API2[&#x22;Search Service Cache&#x22;]
  API3[&#x22;Recommendation Cache&#x22;]

  PS -- &#x22;publish invalidation&#x22; --> EV
  EV -- &#x22;fan-out&#x22; --> API1
  EV -- &#x22;fan-out&#x22; --> API2
  EV -- &#x22;fan-out&#x22; --> API3

  class EV events
  class PS,API1,API2,API3 client"
/>

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

## Implementation [#implementation]

### Cache Invalidation Publisher [#cache-invalidation-publisher]

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

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="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")
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="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()
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="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();
    ```
  </Tab>

  <Tab value="Java">
    ```java title="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();
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="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");
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="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()
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="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;
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="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(())
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="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
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="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)
    ```
  </Tab>
</Tabs>

### Cache Listener [#cache-listener]

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

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="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)
        }),
    )
    ```
  </Tab>

  <Tab value="Python">
    ```python title="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(),
    )
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="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),
    });
    ```
  </Tab>

  <Tab value="Java">
    ```java title="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());
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="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()}");
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="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}") }
    )
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="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;
        }
    );
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="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?;
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="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
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="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
      )
    ```
  </Tab>
</Tabs>

## Production Considerations [#production-considerations]

<Accordions>
  <Accordion title="What if a cache listener misses an event?">
    Events use at-most-once delivery, so a missed invalidation results in stale cache data. Mitigate this by adding a TTL (time-to-live) to all cache entries. Even if an invalidation is missed, the entry expires naturally. For critical data, add periodic full-sync reconciliation.
  </Accordion>

  <Accordion title="Thundering herd on cache miss">
    When all service instances evict the same key simultaneously, they may all attempt to reload from the database at once. Implement a cache stampede lock (only one instance fetches, others wait) or add jitter to the eviction timing.
  </Accordion>

  <Accordion title="Selective invalidation">
    Use hierarchical channels (e.g., `cache.invalidate.products`, `cache.invalidate.users`) so services can subscribe only to entity types they cache. This reduces unnecessary processing using [wildcard subscriptions](/learn/events/tutorials/wildcard-subscriptions).
  </Accordion>
</Accordions>

## Related [#related]

* [Wildcard Subscriptions](/learn/events/tutorials/wildcard-subscriptions) for selective cache listening
* [Multicast Events](/learn/events/tutorials/multicast) for broadcasting to events and queues simultaneously
* [Events Store](/learn/events-store) for guaranteed cache invalidation delivery