Cached Query
Enable server-side response caching for KubeMQ Queries to serve repeated requests from cache using the Go SDK.
Overview
Query response caching lets the broker answer repeat requests without re-running your handler — useful when a query is expensive to compute (a database lookup, an aggregation, a downstream call) but the same input is asked for repeatedly in a short window. Only the first request pays the processing cost; every other caller gets the same answer straight from the broker.
Tag a query with SetCacheKey and SetCacheTTL. The first query with a given key is a miss: it reaches the handler, and the broker stores the response under that key for the TTL. A subsequent query with the same key is a hit — the broker returns the stored response directly without invoking the handler. CacheHit on the response tells you which happened.
Gotchas: the cache is keyed by the string you choose, not by the query body — if the underlying data changes mid-TTL, callers can get a stale answer until it expires. Keys are scoped per channel, so the same key on another channel is a separate entry. Caching only helps when requests genuinely repeat with the same key.
Prerequisites
- KubeMQ server running on
localhost:50000 - Go SDK installed (
go get github.com/kubemq-io/kubemq-go/v2)
Code
// Example: queries/cached-query
//
// Demonstrates query caching with CacheKey and CacheTTL.
// The first query triggers the handler; subsequent queries with the same
// cache key are served from cache (CacheHit=true) without calling the handler.
//
// Channel: go-queries.cached-query
// Client ID: go-queries-cached-query-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/kubemq-io/kubemq-go/v2"
)
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
client, err := kubemq.NewClient(ctx,
kubemq.WithAddress("localhost", 50000),
kubemq.WithClientId("go-queries-cached-query-client"),
)
if err != nil {
log.Fatal(err)
}
defer client.Close()
channel := "go-queries.cached-query"
handlerCalled := make(chan struct{}, 2)
// Register a query handler.
sub, err := client.SubscribeToQueries(ctx, channel, "",
kubemq.WithOnQueryReceive(func(q *kubemq.QueryReceive) {
fmt.Printf("Handler called: body=%s\n", q.Body)
resp := kubemq.NewQueryReply().
SetRequestId(q.Id).
SetResponseTo(q.ResponseTo).
SetBody([]byte("cached-result")).
SetExecutedAt(time.Now())
_ = client.SendQueryResponse(ctx, resp)
select {
case handlerCalled <- struct{}{}:
default:
}
}),
kubemq.WithOnError(func(err error) {
log.Println("Subscription error:", err)
}),
)
if err != nil {
log.Fatal(err)
}
defer sub.Unsubscribe()
time.Sleep(300 * time.Millisecond)
// First query: handler is called, result is cached for 60 seconds.
q1Resp, err := client.SendQuery(ctx, kubemq.NewQuery().
SetChannel(channel).
SetBody([]byte("cacheable-query")).
SetTimeout(10*time.Second).
SetCacheKey("my-cache-key").
SetCacheTTL(60*time.Second))
if err != nil {
log.Fatal(err)
}
fmt.Printf("First query: executed=%v cacheHit=%v body=%s\n",
q1Resp.Executed, q1Resp.CacheHit, q1Resp.Body)
<-handlerCalled
// Second query with same cache key: served from cache (no handler call).
q2Resp, err := client.SendQuery(ctx, kubemq.NewQuery().
SetChannel(channel).
SetBody([]byte("cacheable-query")).
SetTimeout(10*time.Second).
SetCacheKey("my-cache-key").
SetCacheTTL(60*time.Second))
if err != nil {
log.Fatal(err)
}
fmt.Printf("Second query: executed=%v cacheHit=%v body=%s\n",
q2Resp.Executed, q2Resp.CacheHit, q2Resp.Body)
}
How It Works
kubemq.NewQuery().SetCacheKey("my-cache-key").SetCacheTTL(60*time.Second)instructs the broker to cache the first successful response under the given key for 60 seconds.- On the first query, the broker calls the registered handler, returns the result, and stores it in the cache keyed by
"my-cache-key". - On the second query with the same
CacheKey, the broker serves the cached response directly — the handler is not called;q2Resp.CacheHitistrueandq2Resp.Bodycontains the previously cached payload. - Cache entries are per-broker and per-channel; the TTL resets on each cache miss that triggers the handler. The
handlerCalledchannel verifies that only one handler invocation occurred.
Related
Was this page helpful?