KubeMQ
Client SDKsGoHow-to guidesEvents

Multiple Subscribers

Deliver the same KubeMQ Events to multiple subscribers via fan-out using the Go SDK.

Overview

Fan-out delivery lets several independent consumers each get their own copy of every event published on a channel — the pattern behind broadcasting a notification to every connected service or feeding the same stream to a cache invalidator and a metrics collector at once. Reach for it whenever multiple, unrelated pieces of code all need to react to the same event, rather than compete for it.

It works by calling SubscribeToEvents more than once for the same channel while leaving the group argument empty (""). Each call opens its own stream, and the broker treats every subscriber with no group as broadcast: publishing one event delivers it to every open stream — the opposite of a consumer group, where subscribers sharing a group name split events among themselves for load balancing.

Gotchas: Events pub/sub has no durability — a subscriber that hasn't finished subscribing yet, or that disconnects, simply misses events published in that window; there's no redelivery. Mixing a non-empty group into one subscriber on the same channel silently turns broadcast into load-balancing for it. And because delivery is fully concurrent, shared state your callbacks touch needs its own synchronization.

Prerequisites

  • KubeMQ server running on localhost:50000
  • Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

main.go
// Example: events/multiple-subscribers
//
// Demonstrates multiple subscribers receiving the same event.
// Without a consumer group, each subscriber gets every event (fan-out).
//
// Channel: go-events.multiple-subscribers
// Client ID: go-events-multiple-subscribers-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"sync/atomic"
	"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-events-multiple-subscribers-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-events.multiple-subscribers"
	var count atomic.Int32

	// Create two subscribers on the same channel without a group.
	// Both subscribers should receive every event (fan-out).
	sub1, err := client.SubscribeToEvents(ctx, channel, "",
		kubemq.WithOnEvent(func(event *kubemq.Event) {
			fmt.Printf("Subscriber 1 received: body=%s\n", event.Body)
			count.Add(1)
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Sub1 error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub1.Unsubscribe()

	sub2, err := client.SubscribeToEvents(ctx, channel, "",
		kubemq.WithOnEvent(func(event *kubemq.Event) {
			fmt.Printf("Subscriber 2 received: body=%s\n", event.Body)
			count.Add(1)
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Sub2 error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub2.Unsubscribe()

	// Allow subscriptions to fully establish before publishing.
	time.Sleep(time.Second)

	// Publish an event — both subscribers should receive it.
	err = client.SendEvent(ctx, kubemq.NewEvent().
		SetChannel(channel).
		SetBody([]byte("hello to all subscribers")).
		SetMetadata("fan-out"))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("Event published")

	// Wait briefly for both subscribers to receive the event.
	time.Sleep(2 * time.Second)
	fmt.Printf("Total deliveries: %d (expected 2 for fan-out)\n", count.Load())
}

How It Works

  1. Two independent calls to SubscribeToEvents create two separate gRPC streams for go-events.multiple-subscribers, each with an empty group string.
  2. Without a consumer group the broker operates in broadcast (fan-out) mode: every event published on the channel is delivered to both streams simultaneously.
  3. atomic.Int32 provides race-safe counting across the two callback goroutines.
  4. A one-second sleep before publishing ensures both subscriptions are registered on the server side; in production, consider using WithWaitForReady or an explicit Ping before the first send.

Was this page helpful?

Consumer Group

Load-balance KubeMQ Events across multiple subscribers with a consumer group using the Go SDK.

Stream Send

Publish KubeMQ Events at high throughput by streaming sends with the Go SDK.

On this page

OverviewPrerequisitesCodeHow It WorksRelated