---
title: "Publish/subscribe: one event, many consumer groups"
description: "One order-placed event read by billing, shipping and analytics: each consumer group gets every message, and a late group chooses where it starts."
---

> Queen MQ documentation, for AI agents
> Complete self-contained summary of Queen MQ: https://queenmq.com/llms-brief.txt
> Fetch that first when the question is about the product rather than about this page.
> Index of all pages: https://queenmq.com/llms.txt

# Publish/subscribe: one event, many consumer groups

One event, many readers: when an order is placed, billing charges it, shipping prepares it, and analytics counts it. Each reader must get every event, at its own pace, and a reader added later must be able to start from the beginning.

Each reader is a consumer group. A group gets every message of the queue and keeps its own position; the workers inside a group share its work, as in a work queue.

**Figure.** Orders 3001, 3002 and 3003 are published once to the queue order-placed. Three consumer groups subscribe: billing and shipping from the start, and analytics months later with subscriptionMode all. Each group gets all three orders and keeps its own position. A group created later without subscriptionMode all starts at the end and gets nothing of the past.

Published once, read by every group, also by one that joins later.

JavaScript:

```js
const subscribe = (group) => queen.queue('order-placed')
  .group(group)                                   // each group gets every message
  .subscriptionMode('all')
  .each()
  .consume(async (event) => {
    console.log(group, event.data.orderId)
  })

for (const orderId of [3001, 3002, 3003]) {
  await queen.queue('order-placed').partition(`order-${orderId}`)
    .push([{ data: { orderId } }])
}
await Promise.all([subscribe('billing'), subscribe('shipping')])

// Months later, analytics subscribes too, and reads every order ever placed.
await subscribe('analytics')
```

Python:

```python
def subscribe(group):
    async def handle(event):
        print(group, event["data"]["orderId"])

    return (
        queen.queue("order-placed")
        .group(group)                 # each group gets every message
        .subscription_mode("all")
        .each()
        .consume(handle)
    )

for order_id in [3001, 3002, 3003]:
    await queen.queue("order-placed").partition(f"order-{order_id}").push(
        [{"data": {"orderId": order_id}}]
    )
await asyncio.gather(subscribe("billing"), subscribe("shipping"))

# Months later, analytics subscribes too, and reads every order ever placed.
await subscribe("analytics")
```

Go:

```go
subscribe := func(group string) error {
	return client.Queue("order-placed").
		Group(group). // each group gets every message
		SubscriptionMode(queen.SubscriptionModeAll).
		Each().
		Consume(ctx, func(ctx context.Context, event *queen.Message) error {
			fmt.Println(group, event.Data["orderId"])
			return nil
		}).
		Execute(ctx)
}

for _, orderID := range []int{3001, 3002, 3003} {
	_, err := client.Queue("order-placed").Partition(fmt.Sprintf("order-%d", orderID)).
		Push(map[string]interface{}{"orderId": orderID}).
		Execute(ctx)
	if err != nil {
		log.Fatal(err)
	}
}
var subscribers sync.WaitGroup
for _, group := range []string{"billing", "shipping"} {
	subscribers.Add(1)
	go func() {
		defer subscribers.Done()
		if err := subscribe(group); err != nil {
			log.Fatal(err)
		}
	}()
}
subscribers.Wait()

// Months later, analytics subscribes too, and reads every order ever placed.
if err := subscribe("analytics"); err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
let subscribe = |group: &'static str| {
    let events = queen
        .queue("order-placed")
        .group(group)                       // each group gets every message
        .subscription_mode(SubscriptionMode::All);
    async move {
        events
            .consume(move |event: Message| async move {
                println!("{group} {}", event.data["orderId"]);
                Ok::<_, queen_mq::Error>(())
            })
            .await
    }
};

for order_id in [3001, 3002, 3003] {
    queen
        .queue("order-placed")
        .partition(format!("order-{order_id}"))
        .push(json!({ "orderId": order_id }))
        .await?;
}
tokio::try_join!(subscribe("billing"), subscribe("shipping"))?;

// Months later, analytics subscribes too, and reads every order ever placed.
subscribe("analytics").await?;
```

PHP:

```php
$subscribe = fn (string $group) => $queen->queue('order-placed')
    ->group($group)                                 // each group gets every message
    ->subscriptionMode('all')
    ->each()
    ->consume(function (array $event) use ($group) {
        echo $group, ' ', $event['data']['orderId'], "\n";
    })
    ->execute();

foreach ([3001, 3002, 3003] as $orderId) {
    $queen->queue('order-placed')->partition("order-{$orderId}")
        ->push([['data' => ['orderId' => $orderId]]])
        ->execute();
}
$subscribe('billing');           // one after the other here: in production each group
$subscribe('shipping');          // is a worker process of its own

// Months later, analytics subscribes too, and reads every order ever placed.
$subscribe('analytics');
```

C++:

```cpp
auto subscribe = [&](const std::string& group) {
    client.queue("order-placed")
        .group(group)                  // each group gets every message
        .subscription_mode("all")
        .each()
        .consume([&](const json& event) {
            std::cout << group + " " + event["data"]["orderId"].dump() + "\n";
        });
};

for (int order_id : {3001, 3002, 3003}) {
    client.queue("order-placed").partition("order-" + std::to_string(order_id))
        .push({{{"data", {{"orderId", order_id}}}}});
}
std::thread billing(subscribe, "billing"), shipping(subscribe, "shipping");
billing.join(); shipping.join();

// Months later, analytics subscribes too, and reads every order ever placed.
subscribe("analytics");
```

Billing and shipping each got the three orders. Analytics, started later, read all three from the start.

**Why:** the queue is a log: reading does not delete. So a new subscriber costs nothing to the publisher, and a group that falls behind slows down only itself.

One thing to know: a new group starts at the end of the queue unless it asks for `subscriptionMode('all')`, or a point in time with `subscriptionFrom`. Our check proved it: a group created without it found nothing.

Built from: [Pop and consume](/writing-applications/pop/#queue-mode-and-consumer-groups).

Source: https://queenmq.com/patterns/publish-subscribe/index.mdx