# Query Caching (/learn/rpc/tutorials/query-caching)



## What You Will Build [#what-you-will-build]

A cached product lookup — the first query hits the responder, and subsequent queries are served directly from KubeMQ's in-memory cache until the TTL expires.

## How Query Caching Works [#how-query-caching-works]

<Mermaid
  chart="sequenceDiagram
    participant S as Sender
    participant K as KubeMQ
    participant R as Responder

    S->>K: Query (CacheKey=&#x22;product-123&#x22;, CacheTTL=60s)
    K->>K: Cache MISS
    K->>R: Deliver query
    R->>K: Response (body: product data)
    K->>K: Store in cache
    K->>S: Response (CacheHit: false)

    S->>K: Query (CacheKey=&#x22;product-123&#x22;)
    K->>K: Cache HIT
    K->>S: Response (CacheHit: true)
    Note over R: Responder NOT called"
/>

*The first query misses the cache and reaches the responder; KubeMQ stores the reply and serves every later query with the same cache key directly until the TTL expires.*

## Steps [#steps]

<Steps>
  <Step>
    ### Set Up a Query Responder [#set-up-a-query-responder]

    The responder handles product lookups. With caching enabled, it is only called on cache misses.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="product_responder.go"
        package main

        import (
            "context"
            "fmt"
            "log"
            "time"

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

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

            _, err = client.SubscribeToQueries(ctx, "products.lookup", "",
                kubemq.WithOnQueryReceive(func(query *kubemq.QueryReceive) {
                    productId := string(query.Body)
                    fmt.Printf("Cache MISS — fetching product %s from database\n", productId)

                    resp := kubemq.NewQueryReply().
                        SetRequestId(query.Id).
                        SetResponseTo(query.ResponseTo).
                        SetBody([]byte(fmt.Sprintf(
                            `{"productId":"%s","name":"Widget","price":29.99,"stock":150}`,
                            productId))).
                        SetExecutedAt(time.Now())
                    _ = client.SendQueryResponse(ctx, resp)
                }),
                kubemq.WithOnError(func(err error) {
                    log.Println("Error:", err)
                }),
            )
            if err != nil {
                log.Fatal(err)
            }

            fmt.Println("Product responder ready...")
            <-ctx.Done()
        }
        ```
      </Tab>

      <Tab value="Python">
        ```python title="product_responder.py"
        import time
        from kubemq.cq import (
            Client as CQClient, QueriesSubscription,
            QueryReceived, QueryResponse, CancellationToken,
        )

        def on_query(request: QueryReceived) -> None:
            product_id = request.body.decode("utf-8")
            print(f"Cache MISS — fetching product {product_id} from database")
            client.send_response_message(
                QueryResponse(
                    query_received=request,
                    is_executed=True,
                    body=f'{{"productId":"{product_id}","name":"Widget","price":29.99,"stock":150}}'.encode(),
                )
            )

        client = CQClient(address="localhost:50000")
        cancel = CancellationToken()
        client.subscribe_to_queries(
            subscription=QueriesSubscription(
                channel="products.lookup",
                on_receive_query_callback=on_query,
                on_error_callback=lambda e: print(f"Error: {e}"),
            ),
            cancel=cancel,
        )
        print("Product responder ready...")
        time.sleep(3600)
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="product_responder.js"
        const { KubeMQClient } = require("kubemq-js");

        const client = new KubeMQClient({ address: "localhost:50000" });

        client.subscribeToQueries({
          channel: "products.lookup",
          onQuery: (query) => {
            const productId = Buffer.from(query.body).toString();
            console.log(`Cache MISS — fetching product ${productId} from database`);
            client.sendQueryResponse({
              requestId: query.id,
              isExecuted: true,
              body: Buffer.from(JSON.stringify({
                productId, name: "Widget", price: 29.99, stock: 150,
              })),
            });
          },
          onError: (err) => console.error("Error:", err.message),
        });

        console.log("Product responder ready...");
        ```
      </Tab>

      <Tab value="Java">
        ```java title="ProductResponder.java"
        CQClient client = CQClient.builder()
            .address("localhost:50000")
            .clientId("product-responder")
            .build();

        client.subscribeToQueries(QueriesSubscription.builder()
            .channel("products.lookup")
            .onReceiveQueryCallback(query -> {
                String productId = new String(query.getBody());
                System.out.println("Cache MISS — fetching product " + productId);
                return QueryResponseMessage.builder()
                    .requestId(query.getId())
                    .isExecuted(true)
                    .body(String.format(
                        "{\"productId\":\"%s\",\"name\":\"Widget\",\"price\":29.99,\"stock\":150}",
                        productId).getBytes())
                    .build();
            })
            .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
            .build());

        System.out.println("Product responder ready...");
        Thread.sleep(3600000);
        client.close();
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="ProductResponder.cs"
        await using var client = new KubeMQClient(new KubeMQClientOptions());
        await client.ConnectAsync();

        Console.WriteLine("Product responder ready...");
        await foreach (var query in client.SubscribeToQueriesAsync(
            new QueriesSubscription { Channel = "products.lookup" }))
        {
            var productId = Encoding.UTF8.GetString(query.Body.Span);
            Console.WriteLine($"Cache MISS — fetching product {productId}");
            await client.SendQueryResponseAsync(new QueryResponse
            {
                RequestId = query.Id,
                IsExecuted = true,
                Body = Encoding.UTF8.GetBytes(
                    $"{{\"productId\":\"{productId}\",\"name\":\"Widget\",\"price\":29.99,\"stock\":150}}")
            });
        }
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="ProductResponder.kt"
        val client = CQClient("localhost:50000")

        client.subscribeToQueries(
            channel = "products.lookup",
            onQuery = { query ->
                val productId = String(query.body)
                println("Cache MISS — fetching product $productId from database")
                client.sendQueryResponse(
                    requestId = query.id, isExecuted = true,
                    body = """{"productId":"$productId","name":"Widget","price":29.99,"stock":150}""".toByteArray()
                )
            },
            onError = { err -> System.err.println("Error: ${err.message}") }
        )

        println("Product responder ready...")
        Thread.sleep(3600000)
        client.close()
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="product_responder.cpp"
        #include <kubemq/client.h>
        #include <iostream>
        #include <thread>

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

        client.subscribeToQueries("products.lookup", "",
            [&client](const kubemq::QueryReceive& query) {
                std::cout << "Cache MISS — fetching product " << query.body << std::endl;
                std::string data = R"({"productId":")" + query.body +
                    R"(","name":"Widget","price":29.99,"stock":150})";
                client.sendQueryResponse(query.id, true, data);
            },
            [](const std::string& err) {
                std::cerr << "Error: " << err << std::endl;
            }
        );

        std::cout << "Product responder ready..." << std::endl;
        std::this_thread::sleep_for(std::chrono::hours(1));
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="product_responder.rs"
        use kubemq::prelude::*;
        use kubemq::QueryReplyBuilder;
        use std::time::Duration;

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

            let responder = client.clone();
            let sub = client
                .subscribe_to_queries(
                    "products.lookup",
                    "",
                    move |query| {
                        let rc = responder.clone();
                        Box::pin(async move {
                            let product_id = String::from_utf8_lossy(&query.body).to_string();
                            println!("Cache MISS — fetching product {product_id} from database");

                            let reply = QueryReplyBuilder::new()
                                .request_id(&query.id)
                                .response_to(&query.response_to)
                                .body(
                                    format!(
                                        r#"{{"productId":"{product_id}","name":"Widget","price":29.99,"stock":150}}"#
                                    )
                                    .into_bytes(),
                                )
                                .build();
                            tokio::spawn(async move {
                                let _ = rc.send_query_response(reply).await;
                            });
                        })
                    },
                    None,
                )
                .await?;

            println!("Product responder ready...");
            tokio::signal::ctrl_c().await.ok();
            sub.unsubscribe().await;
            client.close().await
        }
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="product_responder.rb"
        require 'kubemq'

        client = KubeMQ::CQClient.new(address: 'localhost:50000', client_id: 'product-responder')
        cancel = KubeMQ::CancellationToken.new

        sub = KubeMQ::CQ::QueriesSubscription.new(channel: 'products.lookup')
        client.subscribe_to_queries(sub, cancellation_token: cancel, on_error: lambda { |e|
          puts "Error: #{e.message}"
        }) do |query|
          product_id = query.body
          puts "Cache MISS — fetching product #{product_id} from database"
          response = KubeMQ::CQ::QueryResponseMessage.new(
            request_id: query.id,
            reply_channel: query.reply_channel,
            executed: true,
            body: "{\"productId\":\"#{product_id}\",\"name\":\"Widget\",\"price\":29.99,\"stock\":150}"
          )
          client.send_response(response)
        end

        puts 'Product responder ready...'
        cancel.wait
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="product_responder.exs"
        {:ok, client} =
          KubeMQ.Client.start_link(address: "localhost:50000", client_id: "product-responder")

        {:ok, _sub} =
          KubeMQ.Client.subscribe_to_queries(client, "products.lookup",
            on_query: fn query ->
              product_id = query.body
              IO.puts("Cache MISS — fetching product #{product_id} from database")

              KubeMQ.QueryReply.new(
                request_id: query.id,
                response_to: query.reply_channel,
                executed: true,
                body: ~s({"productId":"#{product_id}","name":"Widget","price":29.99,"stock":150})
              )
            end,
            on_error: fn err -> IO.puts("Error: #{err.message}") end
          )

        IO.puts("Product responder ready...")
        Process.sleep(:infinity)
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Send a Query with CacheKey [#send-a-query-with-cachekey]

    Set `CacheKey` and `CacheTTL` on the query request to enable caching. The first query is a cache miss, and subsequent queries within the TTL are served from cache.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="cached_query.go"
        resp, err := client.SendQuery(ctx, kubemq.NewQuery().
            SetChannel("products.lookup").
            SetBody([]byte("PROD-123")).
            SetTimeout(10 * time.Second).
            SetCacheKey("product-PROD-123").
            SetCacheTTL(60 * time.Second))
        if err != nil {
            log.Fatal(err)
        }
        log.Printf("Product: %s (cache hit: %v)", resp.Body, resp.CacheHit)
        ```
      </Tab>

      <Tab value="Python">
        ```python title="cached_query.py"
        from kubemq.cq import Client as CQClient, QueryMessage

        with CQClient(address="localhost:50000") as client:
            response = client.send_query(
                QueryMessage(
                    channel="products.lookup",
                    body=b"PROD-123",
                    timeout_in_seconds=10,
                    cache_key="product-PROD-123",
                    cache_ttl_in_seconds=60,
                )
            )
            print(f"Product: {response.body.decode('utf-8')} (cache hit: {response.cache_hit})")
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="cached_query.js"
        const response = await client.sendQuery({
          channel: "products.lookup",
          body: Buffer.from("PROD-123"),
          timeoutInSeconds: 10,
          cacheKey: "product-PROD-123",
          cacheTTL: 60000,
        });
        console.log("Product:", Buffer.from(response.body).toString(),
          "(cache hit:", response.cacheHit, ")");
        ```
      </Tab>

      <Tab value="Java">
        ```java title="CachedQuery.java"
        QueryResponseMessage response = client.sendQueryRequest(
            QueryMessage.builder()
                .channel("products.lookup")
                .body("PROD-123".getBytes())
                .timeout(10000)
                .cacheKey("product-PROD-123")
                .cacheTTL(60000)
                .build());
        System.out.println("Product: " + new String(response.getBody())
            + " (cache hit: " + response.isCacheHit() + ")");
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="CachedQuery.cs"
        var response = await client.SendQueryAsync(new QueryMessage
        {
            Channel = "products.lookup",
            Body = Encoding.UTF8.GetBytes("PROD-123"),
            Timeout = TimeSpan.FromSeconds(10),
            CacheKey = "product-PROD-123",
            CacheTTL = TimeSpan.FromSeconds(60)
        });
        Console.WriteLine($"Product: {Encoding.UTF8.GetString(response.Body.Span)}"
            + $" (cache hit: {response.CacheHit})");
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="CachedQuery.kt"
        val response = client.sendQuery(QueryMessage(
            channel = "products.lookup",
            body = "PROD-123".toByteArray(),
            timeout = 10000,
            cacheKey = "product-PROD-123",
            cacheTTL = 60000
        ))
        println("Product: ${String(response.body)} (cache hit: ${response.cacheHit})")
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="cached_query.cpp"
        kubemq::QueryMessage query;
        query.channel = "products.lookup";
        query.body = "PROD-123";
        query.timeout = 10000;
        query.cacheKey = "product-PROD-123";
        query.cacheTTL = 60000;

        auto response = client.sendQuery(query);
        std::cout << "Product: " << response.body
                  << " (cache hit: " << response.cacheHit << ")" << std::endl;
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="cached_query.rs"
        use kubemq::QueryBuilder;
        use std::time::Duration;

        let query = QueryBuilder::new()
            .channel("products.lookup")
            .body(b"PROD-123".to_vec())
            .timeout(Duration::from_secs(10))
            .cache_key("product-PROD-123")
            .cache_ttl(Duration::from_secs(60))
            .build();

        let response = client.send_query(query).await?;
        println!(
            "Product: {} (cache hit: {})",
            String::from_utf8_lossy(&response.body),
            response.cache_hit
        );
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="cached_query.rb"
        msg = KubeMQ::CQ::QueryMessage.new(
          channel: 'products.lookup',
          body: 'PROD-123',
          timeout: 10,
          cache_key: 'product-PROD-123',
          cache_ttl: 60
        )

        response = client.send_query(msg)
        puts "Product: #{response.body} (cache hit: #{response.cache_hit})"
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="cached_query.exs"
        query =
          KubeMQ.Query.new(
            channel: "products.lookup",
            body: "PROD-123",
            timeout: 10_000,
            cache_key: "product-PROD-123",
            cache_ttl: 60_000
          )

        case KubeMQ.Client.send_query(client, query) do
          {:ok, resp} ->
            IO.puts("Product: #{resp.body} (cache hit: #{resp.cache_hit})")

          {:error, err} ->
            IO.puts("Error: #{err.message}")
        end
        ```
      </Tab>
    </Tabs>
  </Step>

  <Step>
    ### Verify Cache Behavior [#verify-cache-behavior]

    Send the same query twice to observe the cache in action.

    <Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
      <Tab value="Go">
        ```go title="verify_cache.go"
        query := kubemq.NewQuery().
            SetChannel("products.lookup").
            SetBody([]byte("PROD-123")).
            SetTimeout(10 * time.Second).
            SetCacheKey("product-PROD-123").
            SetCacheTTL(60 * time.Second)

        resp1, _ := client.SendQuery(ctx, query)
        log.Printf("First query  — cache hit: %v", resp1.CacheHit) // false

        resp2, _ := client.SendQuery(ctx, query)
        log.Printf("Second query — cache hit: %v", resp2.CacheHit) // true
        ```
      </Tab>

      <Tab value="Python">
        ```python title="verify_cache.py"
        query = QueryMessage(
            channel="products.lookup",
            body=b"PROD-123",
            timeout_in_seconds=10,
            cache_key="product-PROD-123",
            cache_ttl_in_seconds=60,
        )

        resp1 = client.send_query(query)
        print(f"First query  — cache hit: {resp1.cache_hit}")  # False

        resp2 = client.send_query(query)
        print(f"Second query — cache hit: {resp2.cache_hit}")  # True
        ```
      </Tab>

      <Tab value="Node.js">
        ```javascript title="verify_cache.js"
        const queryOpts = {
          channel: "products.lookup",
          body: Buffer.from("PROD-123"),
          timeoutInSeconds: 10,
          cacheKey: "product-PROD-123",
          cacheTTL: 60000,
        };

        const resp1 = await client.sendQuery(queryOpts);
        console.log("First query  — cache hit:", resp1.cacheHit); // false

        const resp2 = await client.sendQuery(queryOpts);
        console.log("Second query — cache hit:", resp2.cacheHit); // true
        ```
      </Tab>

      <Tab value="Java">
        ```java title="VerifyCache.java"
        QueryMessage query = QueryMessage.builder()
            .channel("products.lookup")
            .body("PROD-123".getBytes())
            .timeout(10000)
            .cacheKey("product-PROD-123")
            .cacheTTL(60000)
            .build();

        var resp1 = client.sendQueryRequest(query);
        System.out.println("First query  — cache hit: " + resp1.isCacheHit()); // false

        var resp2 = client.sendQueryRequest(query);
        System.out.println("Second query — cache hit: " + resp2.isCacheHit()); // true
        ```
      </Tab>

      <Tab value="C#">
        ```csharp title="VerifyCache.cs"
        var query = new QueryMessage
        {
            Channel = "products.lookup",
            Body = Encoding.UTF8.GetBytes("PROD-123"),
            Timeout = TimeSpan.FromSeconds(10),
            CacheKey = "product-PROD-123",
            CacheTTL = TimeSpan.FromSeconds(60)
        };

        var resp1 = await client.SendQueryAsync(query);
        Console.WriteLine($"First query  — cache hit: {resp1.CacheHit}"); // false

        var resp2 = await client.SendQueryAsync(query);
        Console.WriteLine($"Second query — cache hit: {resp2.CacheHit}"); // true
        ```
      </Tab>

      <Tab value="Kotlin">
        ```kotlin title="VerifyCache.kt"
        val query = QueryMessage(
            channel = "products.lookup",
            body = "PROD-123".toByteArray(),
            timeout = 10000,
            cacheKey = "product-PROD-123",
            cacheTTL = 60000
        )

        val resp1 = client.sendQuery(query)
        println("First query  — cache hit: ${resp1.cacheHit}") // false

        val resp2 = client.sendQuery(query)
        println("Second query — cache hit: ${resp2.cacheHit}") // true
        ```
      </Tab>

      <Tab value="C++">
        ```cpp title="verify_cache.cpp"
        kubemq::QueryMessage query;
        query.channel = "products.lookup";
        query.body = "PROD-123";
        query.timeout = 10000;
        query.cacheKey = "product-PROD-123";
        query.cacheTTL = 60000;

        auto resp1 = client.sendQuery(query);
        std::cout << "First query  — cache hit: " << resp1.cacheHit << std::endl; // 0

        auto resp2 = client.sendQuery(query);
        std::cout << "Second query — cache hit: " << resp2.cacheHit << std::endl; // 1
        ```
      </Tab>

      <Tab value="Rust">
        ```rust title="verify_cache.rs"
        use kubemq::QueryBuilder;
        use std::time::Duration;

        let build_query = || {
            QueryBuilder::new()
                .channel("products.lookup")
                .body(b"PROD-123".to_vec())
                .timeout(Duration::from_secs(10))
                .cache_key("product-PROD-123")
                .cache_ttl(Duration::from_secs(60))
                .build()
        };

        let resp1 = client.send_query(build_query()).await?;
        println!("First query  — cache hit: {}", resp1.cache_hit); // false

        let resp2 = client.send_query(build_query()).await?;
        println!("Second query — cache hit: {}", resp2.cache_hit); // true
        ```
      </Tab>

      <Tab value="Ruby">
        ```ruby title="verify_cache.rb"
        msg = KubeMQ::CQ::QueryMessage.new(
          channel: 'products.lookup',
          body: 'PROD-123',
          timeout: 10,
          cache_key: 'product-PROD-123',
          cache_ttl: 60
        )

        resp1 = client.send_query(msg)
        puts "First query  — cache hit: #{resp1.cache_hit}"  # false

        resp2 = client.send_query(msg)
        puts "Second query — cache hit: #{resp2.cache_hit}"  # true
        ```
      </Tab>

      <Tab value="Elixir">
        ```elixir title="verify_cache.exs"
        query =
          KubeMQ.Query.new(
            channel: "products.lookup",
            body: "PROD-123",
            timeout: 10_000,
            cache_key: "product-PROD-123",
            cache_ttl: 60_000
          )

        {:ok, resp1} = KubeMQ.Client.send_query(client, query)
        IO.puts("First query  — cache hit: #{resp1.cache_hit}")  # false

        {:ok, resp2} = KubeMQ.Client.send_query(client, query)
        IO.puts("Second query — cache hit: #{resp2.cache_hit}")  # true
        ```
      </Tab>
    </Tabs>
  </Step>
</Steps>

## Cache Configuration [#cache-configuration]

| Setting              | Description                                                               |
| -------------------- | ------------------------------------------------------------------------- |
| `CacheKey`           | Any string identifying the cached response (use `entity-type-id` pattern) |
| `CacheTTL`           | Time-to-live in milliseconds. Required when `CacheKey` is set.            |
| **Storage**          | In-memory on the KubeMQ server                                            |
| **Cleanup interval** | Every 10 seconds                                                          |
| **Persistence**      | None — cache is cleared on server restart                                 |

<Callout type="warn">
  Caching is only available for **Queries**. Commands do not support caching because they represent write operations.
</Callout>

## When to Use Caching [#when-to-use-caching]

| Use Case                | Cache?     | Why                            |
| ----------------------- | ---------- | ------------------------------ |
| Product catalog lookups | ✅ Yes      | Data changes infrequently      |
| Configuration values    | ✅ Yes      | Rarely updated                 |
| Exchange rates          | ✅ Yes      | Update every few minutes       |
| User-specific data      | ⚠️ Depends | Use unique cache keys per user |
| Real-time stock prices  | ❌ No       | Data is stale immediately      |
| Authentication tokens   | ❌ No       | Security-sensitive             |

## Next Steps [#next-steps]

<Cards>
  <Card title="Request-Reply Roundtrip" href="/learn/rpc/tutorials/request-reply-roundtrip" description="Complete sender + responder example." />

  <Card title="Reference" href="/learn/rpc/reference" description="Full caching configuration details." />
</Cards>