---
title: "Pop and consume messages with consumer groups"
description: "Pop a batch or run a consume loop: queue mode and consumer groups, where a new group starts, batches, leases and renewal, conflation, seek, and reads by offset."
---

> 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

# Pop and consume messages with consumer groups

There are two ways to read. `consume()` is a loop: it takes messages, calls your handler, acks them when the handler returns, and goes on. `pop()` is one call: it hands you the messages and leaves the ack to you.

**Why:** most workers are loops, and `consume()` writes the loop for you, with the error handling done right. `pop()` is for the cases a loop does not fit: a scheduled job that reads once, a script, or code that wants to decide itself when to ack.

**Figure.** The consume loop. It pops a batch, which leases the partition; calls your handler; acks when the handler returns, or fails the message when it throws; and goes back to pop. With autoAck off, your handler acks itself, often inside a transaction. pop() is one turn of the loop, done by hand.

What consume() does for you.

## Queue mode and consumer groups

Without a consumer group, each message goes to one worker, and the broker sizes every pop for you.

JavaScript:

```js
await queen.queue('orders')
  .each()                         // the handler gets one message at a time
  .limit(1)                       // stop after one message; a worker leaves this out
  .consume(async (msg) => {       // acked when the handler returns, retried if it throws
    console.log(msg.data)
  })
```

Python:

```python
async def handle(msg):                 # acked when it returns, retried if it raises
    print(msg["data"])

await (
    queen.queue("orders")
    .each()                            # the handler gets one message at a time
    .limit(1)                          # stop after one message; a worker leaves this out
    .consume(handle)
)
```

Go:

```go
err = client.Queue("orders").
	Each().   // the handler gets one message at a time
	Limit(1). // stop after one message; a worker leaves this out
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		fmt.Println(msg.Data) // acked when this returns nil, retried on an error
		return nil
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
use queen_mq::Message;

queen
    .queue("orders")
    .limit(1) // stop after one message; a worker leaves this out
    .consume(|msg: Message| async move {
        println!("{}", msg.data); // acked when this returns Ok, retried on an Err
        Ok::<_, queen_mq::Error>(())
    })
    .await?;
```

PHP:

```php
$queen->queue('orders')
    ->each()                          // the handler gets one message at a time
    ->limit(1)                        // stop after one message; a worker leaves this out
    ->consume(function (array $msg) { // acked when it returns, retried if it throws
        print_r($msg['data']);
    })
    ->execute();
```

C++:

```cpp
client.queue("orders")
    .each()                        // the handler gets one message at a time
    .limit(1)                      // stop after one message; a worker leaves this out
    .consume([](const json& msg) { // acked when it returns, retried if it throws
        std::cout << msg["data"] << std::endl;
    });
```

With a consumer group, every group gets every message and the workers of one group share them. A new group starts at the tail, so this one asks for everything already in the queue. Here every option is set by hand, and the handler acks.

JavaScript:

```js
await queen.queue('orders')
  .group('billing')               // every group gets every message
  .subscriptionMode('all')        // a new group starts at the tail unless told otherwise
  .batch(100)                     // up to 100 messages per pop...
  .partitions(10)                 // ...from up to 10 partitions
  .renewLease(true, 2000)         // heartbeat every 2 s while the handler runs
  .autoAck(false)                 // the handler decides the outcome
  .each()
  .limit(1)
  .consume(async (msg) => {
    const ok = msg.data.total > 0
    await queen.ack(msg, ok ? 'completed' : 'failed', { group: 'billing' })
  })
```

Python:

```python
async def bill(msg):
    ok = msg["data"]["total"] > 0
    await queen.ack(msg, "completed" if ok else "failed", {"group": "billing"})

await (
    queen.queue("orders")
    .group("billing")                  # every group gets every message
    .subscription_mode("all")          # a new group starts at the tail unless told otherwise
    .batch(100)                        # up to 100 messages per pop...
    .partitions(10)                    # ...from up to 10 partitions
    .renew_lease(True, 2000)           # heartbeat every 2 s while the handler runs
    .auto_ack(False)                   # the handler decides the outcome
    .each()
    .limit(1)
    .consume(bill)
)
```

Go:

```go
err = client.Queue("orders").
	Group("billing").                            // every group gets every message
	SubscriptionMode(queen.SubscriptionModeAll). // a new group starts at the tail unless told otherwise
	Batch(100).                                  // up to 100 messages per pop...
	Partitions(10).                              // ...from up to 10 partitions
	RenewLease(true, 2000).                      // heartbeat every 2 s while the handler runs
	AutoAck(false).                              // the handler decides the outcome
	Each().
	Limit(1).
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		ok := msg.Data["total"].(float64) > 0
		_, err := client.Ack(ctx, msg, ok, queen.AckOptions{ConsumerGroup: "billing"})
		return err
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
use queen_mq::{AckStatus, SubscriptionMode};

let client = queen.clone();
queen
    .queue("orders")
    .group("billing")                     // every group gets every message
    .subscription_mode(SubscriptionMode::All) // a new group starts at the tail unless told otherwise
    .batch(100)                           // up to 100 messages per pop...
    .partitions(10)                       // ...from up to 10 partitions
    .renew_lease(Duration::from_secs(2))  // heartbeat every 2 s while the handler runs
    .auto_ack(false)                      // the handler decides the outcome
    .limit(1)
    .consume(move |msg: Message| {
        let client = client.clone();
        async move {
            let ok = msg.data["total"].as_f64().unwrap_or(0.0) > 0.0;
            let status = if ok { AckStatus::Completed } else { AckStatus::Failed };
            client.ack_with(&msg, status, None).await?;
            Ok::<_, queen_mq::Error>(())
        }
    })
    .await?;
```

PHP:

```php
$queen->queue('orders')
    ->group('billing')                // every group gets every message
    ->subscriptionMode('all')         // a new group starts at the tail unless told otherwise
    ->batch(100)                      // up to 100 messages per pop...
    ->partitions(10)                  // ...from up to 10 partitions
    ->renewLease(true, 2000)          // renew every 2 s, between messages
    ->autoAck(false)                  // the handler decides the outcome
    ->each()
    ->limit(1)
    ->consume(function (array $msg) use ($queen) {
        $ok = $msg['data']['total'] > 0;
        $queen->ack($msg, $ok ? 'completed' : 'failed', ['group' => 'billing']);
    })
    ->execute();
```

C++:

```cpp
client.queue("orders")
    .group("billing")          // every group gets every message
    .subscription_mode("all")  // a new group starts at the tail unless told otherwise
    .batch(100)                // up to 100 messages per pop...
    .partitions(10)            // ...from up to 10 partitions
    .renew_lease(true, 2000)   // heartbeat every 2 s while the handler runs
    .auto_ack(false)           // the handler decides the outcome
    .each()
    .limit(1)
    .consume([&](const json& msg) {
        bool ok = msg["data"]["total"].get<double>() > 0;
        client.ack(msg, ok, {{"group", "billing"}});  // true: completed, false: failed
    });
```

Ack `failed` and the message comes back, up to the queue's `retryLimit`; after the last retry it goes to the dead-letter queue.

**Why:** one queue often feeds several services: billing, emails and analytics all want every order. Each service is a consumer group, reading at its own pace, and none of them can slow the others down. Inside a group, the workers compete, so you add workers to go faster.

**Figure.** Left, queue mode with no consumer group: the partitions of orders are shared out, and each message goes to exactly one of the workers w1, w2 and w3. Right, three consumer groups, billing, emails and analytics, on the same queue: each group gets every message and keeps its own position; inside billing, two workers split the partitions between them.

Queue mode shares the messages; every consumer group gets all of them.

## One message at a time, or a whole batch

With `.each()`, the handler gets one message at a time. Without it, the handler gets the whole batch as an array.

JavaScript:

```js
await queen.queue('events')
  .group('archiver')
  .subscriptionMode('all')
  .batch(100)                         // up to 100 messages per call
  .limit(1000)
  .consume(async (messages) => {      // no .each(): the handler gets the whole batch
    console.log('archiving', messages.length)
  })                                  // all acked if it returns, all nacked if it throws
```

Python:

```python
async def archive(messages):          # no .each(): the handler gets the whole batch
    print("archiving", len(messages))

await (
    queen.queue("events")
    .group("archiver")
    .subscription_mode("all")
    .batch(100)                       # up to 100 messages per call
    .limit(1000)
    .consume(archive)                 # all acked if it returns, all nacked if it raises
)
```

Go:

```go
err = client.Queue("events").
	Group("archiver").
	SubscriptionMode(queen.SubscriptionModeAll).
	Batch(100). // up to 100 messages per call
	Limit(1000).
	// ConsumeBatch, not Consume: the handler gets the whole batch
	ConsumeBatch(ctx, func(ctx context.Context, messages []*queen.Message) error {
		fmt.Println("archiving", len(messages))
		return nil // all acked if it returns nil, all nacked if it returns an error
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
queen
    .queue("events")
    .group("archiver")
    .subscription_mode(SubscriptionMode::All)
    .batch(100)                             // up to 100 messages per call
    .limit(1000)
    .consume_batch(|messages: Vec<Message>| async move {
        println!("archiving {}", messages.len()); // the handler gets the whole batch
        Ok::<_, queen_mq::Error>(())
    })                                      // all acked on Ok, all nacked on an Err
    .await?;
```

PHP:

```php
$queen->queue('events')
    ->group('archiver')
    ->subscriptionMode('all')
    ->batch(100)                         // up to 100 messages per call
    ->limit(1000)
    ->consume(function (array $messages) { // no ->each(): the handler gets the whole batch
        echo 'archiving ', count($messages), "\n";
    })                                   // all acked if it returns, all nacked if it throws
    ->execute();
```

C++:

```cpp
client.queue("events")
    .group("archiver")
    .subscription_mode("all")
    .batch(100)                          // up to 100 messages per call
    .limit(1000)
    .consume([&](const json& messages) { // no each(): the handler gets the whole batch
        std::cout << "archiving " << messages.size() << "\n";
    });                                  // all acked if it returns, nacked if it throws
```

What the client does with your handler's result (`autoAck`, on by default):

- **It returns:** the message (or the whole batch) is acked.
- **It throws:** the message (or the whole batch) is nacked as `failed`, which spends one retry, and the loop goes on with the next messages.
- **It throws under `.each()`:** the rest of that partition's messages in the same batch are skipped, because they will be delivered again after the failed one, in order. Messages of other partitions in the batch are handled as usual.
- **With `.autoAck(false)`:** the handler acks itself (in a transaction, for example). A handler that throws is still nacked, so its partition is not stuck until the lease runs out.

**In the other SDKs:** the list above is what JavaScript, Python and Go do. Under `autoAck(false)`, Rust and C++ do not nack a handler that fails (its messages wait for their lease), and PHP stops the loop: catch the error and ack `failed` yourself.

**Why:** a batch is cheaper (one ack for a hundred messages, one database write for a hundred rows), and one at a time is simpler and fails one message instead of a hundred. Pick per worker.

If you add `.onSuccess()` or `.onError()` to a consumer, the client stops acking and nacking for you: you decide in those callbacks what happens to each message. Python and PHP have the same callbacks (PHP with `autoAck(false)`); Go, Rust and C++ have none, so their code turns `autoAck` off and acks in the handler.

JavaScript:

```js
await queen.queue('emails')
  .group('mailer')
  .subscriptionMode('all')
  .each()
  .limit(1)
  .consume(async (msg) => sendEmail(msg.data))
  .onSuccess((msg) => queen.ack(msg, 'completed', { group: 'mailer' }))
  .onError((msg, error) => queen.ack(msg, error.permanent ? 'dlq' : 'failed',
    { group: 'mailer', error: error.message }))
```

Python:

```python
async def deliver(msg):
    await send_email(msg["data"])

async def completed(msg, result):
    await queen.ack(msg, "completed", {"group": "mailer"})

async def failed(msg, error):
    status = "dlq" if getattr(error, "permanent", False) else "failed"
    await queen.ack(msg, status, {"group": "mailer", "error": str(error)})

await (
    queen.queue("emails")
    .group("mailer")
    .subscription_mode("all")
    .each()
    .limit(1)
    .consume(deliver)
    .on_success(completed)
    .on_error(failed)
)
```

Go:

```go
err = client.Queue("emails").
	Group("mailer").
	SubscriptionMode(queen.SubscriptionModeAll).
	AutoAck(false). // the client stops acking: the handler gives every answer
	Each().
	Limit(1).
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		opts := queen.AckOptions{ConsumerGroup: "mailer"}
		err := sendEmail(msg.Data)
		switch {
		case err == nil:
			_, err = client.Ack(ctx, msg, true, opts)
		case errors.Is(err, errPermanent): // trying again cannot help
			// Ack takes no dlq status: a transaction carries it, without the error text.
			_, err = client.Transaction().Ack(msg, "dlq", opts).Commit(ctx)
		default:
			opts.Error = err.Error()
			_, err = client.Ack(ctx, msg, false, opts) // failed: back after retryDelay
		}
		return err
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
let client = queen.clone();
queen
    .queue("emails")
    .group("mailer")
    .subscription_mode(SubscriptionMode::All)
    .limit(1)
    .auto_ack(false) // no callbacks in the Rust client: the handler settles each message
    .consume(move |msg: Message| {
        let client = client.clone();
        async move {
            let (status, error) = match send_email(&msg.data).await {
                Ok(()) => (AckStatus::Completed, None),
                Err(e) if e.permanent => (AckStatus::Dlq, Some(e.message)),
                Err(e) => (AckStatus::Failed, Some(e.message)),
            };
            client.ack_with(&msg, status, error).await?;
            Ok::<_, queen_mq::Error>(())
        }
    })
    .await?;
```

PHP:

```php
$queen->queue('emails')
    ->group('mailer')
    ->subscriptionMode('all')
    ->autoAck(false)                     // PHP: the callbacks below ack, not the loop
    ->each()
    ->limit(1)
    ->consume(fn (array $msg) => sendEmail($msg['data']))
    ->onSuccess(fn (array $msg) => $queen->ack($msg, 'completed', ['group' => 'mailer']))
    ->onError(fn (array $msg, Throwable $error) => $queen->ack($msg,
        $error instanceof PermanentError ? 'dlq' : 'failed',
        ['group' => 'mailer', 'error' => $error->getMessage()]))
    ->execute();                         // with autoAck(false), a throw also ends execute()
```

C++:

```cpp
client.queue("emails")
    .group("mailer")
    .subscription_mode("all")
    .auto_ack(false)                        // the handler settles each message itself
    .each()
    .limit(1)
    .consume([&](const json& msg) {
        try {
            send_email(msg["data"]);
            client.ack(msg, true, {{"group", "mailer"}});
        } catch (const PermanentError& e) { // trying again cannot help
            // QueenClient::ack sends completed or failed only, so the dlq answer
            // goes to the broker's ack route through the client's HTTP layer.
            client.get_http_client()->post("/api/v1/ack", {
                {"transactionId", msg["transactionId"]}, {"leaseId", msg["leaseId"]},
                {"partitionId", msg["partitionId"]}, {"consumerGroup", "mailer"},
                {"status", "dlq"}, {"error", e.what()}});
        } catch (const std::exception& e) { // may work later: spend a retry
            client.ack(msg, false, {{"group", "mailer"}, {"error", e.what()}});
        }
    });
```

**Why:** some failures should be retried and some never will succeed (an email address that does not exist). The callbacks let you send the first kind back with `failed` and the second kind straight to the dead-letter queue with `dlq`.

## How many messages a pop takes: autopilot

By default the broker decides how many messages each pop gets (`batch`, from 100 to 1,000, based on how fast the group acks) and from how many partitions (`partitions`, the ones that have work, shared among the waiting workers). A value you set yourself is never changed. `partitions(N)` lets one pop take messages from up to `N` partitions (1 to 64), with `batch` as the total. `.autopilot(false)` goes back to one message from one partition per pop.

**Why:** the right batch size depends on your handler and your load, and it changes during the day. The broker sees both, so it can size every pop better than a number fixed in the code.

## Leases and long work

A pop leases its messages to the worker for the queue's `leaseTime`. Work that can take longer renews the lease while it runs: `.renewLease(true, 10000)` renews every 10 seconds, and `queen.renew(message)` renews once. In PHP, the renewal happens between messages, so a single long handler calls the renew callback it is given. A lease that has already expired cannot be renewed: its messages go to the next worker, marked as one delivery attempt more.

**Why:** the lease is what protects order and stops two workers from doing the same work. A short lease gets a crashed worker's messages moving again quickly; renewing lets a slow but healthy worker keep them.

## Where a new group starts

A group's starting point is decided on its first pop and then stored: `new` (the default) reads only what is pushed after that first pop, `all` reads everything still kept, and `subscriptionFrom` reads from a point in time. A group that already exists moves with a seek, to a time or to the end.

**Figure.** A partition with messages at offsets 0 to 7, pushed before a new group's first pop. subscriptionMode all starts the group before offset 0 and reads everything. subscriptionFrom a time starts it at the first message from that time, offset 4. The default, new, starts after offset 7 and skips all of them.

Where a new group starts, fixed at its first pop.

JavaScript:

```js
await queen.admin.seekConsumerGroup('archiver', 'events', {
  timestamp: '2026-10-22T00:00:00Z',
})
// the archiver group now reads again everything pushed since that time

await queen.admin.seekConsumerGroup('archiver', 'events', { toEnd: true })
// and now it skips everything it has not read yet
```

Python:

```python
await queen.admin.seek_consumer_group("archiver", "events", {
    "timestamp": "2026-10-22T00:00:00Z",
})
# the archiver group now reads again everything pushed since that time

await queen.admin.seek_consumer_group("archiver", "events", {"toEnd": True})
# and now it skips everything it has not read yet
```

Go:

```go
_, err = client.Admin().SeekConsumerGroup(ctx, "archiver", "events",
	queen.SeekConsumerGroupOptions{Timestamp: "2026-10-22T00:00:00Z"})
if err != nil {
	log.Fatal(err)
}
// the archiver group now reads again everything pushed since that time

_, err = client.Admin().SeekConsumerGroup(ctx, "archiver", "events",
	queen.SeekConsumerGroupOptions{ToEnd: true})
if err != nil {
	log.Fatal(err)
}
// and now it skips everything it has not read yet
```

Rust:

```rust
use queen_mq::SeekRequest;

let since = SeekRequest {
    timestamp: Some("2026-10-22T00:00:00Z".into()),
    position: None,
};
queen.admin().seek_consumer_group("archiver", "events", &since).await?;
// the archiver group now reads again everything pushed since that time

// SeekRequest has no toEnd in the Rust client: a time after the newest
// message moves the cursor to the same place.
let to_end = SeekRequest {
    timestamp: Some("9999-12-31T00:00:00Z".into()),
    position: None,
};
queen.admin().seek_consumer_group("archiver", "events", &to_end).await?;
// and now it skips everything it has not read yet
```

PHP:

```php
$queen->admin()->seekConsumerGroup('archiver', 'events', [
    'timestamp' => '2026-10-22T00:00:00Z',
]);
// the archiver group now reads again everything pushed since that time

$queen->admin()->seekConsumerGroup('archiver', 'events', ['toEnd' => true]);
// and now it skips everything it has not read yet
```

C++:

```cpp
// The C++ client has no admin calls: a seek is a POST to the broker's
// consumer-group route, through the client's own HTTP layer.
auto http = client.get_http_client();
http->post("/api/v1/consumer-groups/archiver/queues/events/seek",
           {{"timestamp", "2026-10-22T00:00:00Z"}});
// the archiver group now reads again everything pushed since that time

http->post("/api/v1/consumer-groups/archiver/queues/events/seek", {{"toEnd", true}});
// and now it skips everything it has not read yet
```

**Why:** a new service usually wants only new events, a new projection wants the whole history, and a bug fix wants to replay the last day. A seek to the end is how a group skips a backlog it no longer needs.

## Only the newest message: conflation

A group with `.conflation()` gets only the newest message of each partition, and its cursor moves past the older ones.

JavaScript:

```js
const [latest] = await queen.queue('prices')
  .group('repricer')
  .subscriptionMode('all')
  .conflation()                       // only the newest message of each partition
  .wait(false)
  .pop()
console.log(latest.data)              // { sku: 'A-1', price: 12 }: 10 and 11 are skipped
await queen.ack(latest, 'completed', { group: 'repricer' })
```

Python:

```python
[latest] = await (
    queen.queue("prices")
    .group("repricer")
    .subscription_mode("all")
    .conflation()                     # only the newest message of each partition
    .wait(False)
    .pop()
)
print(latest["data"])                 # {'sku': 'A-1', 'price': 12}: 10 and 11 are skipped
await queen.ack(latest, "completed", {"group": "repricer"})
```

Go:

```go
newest, err := client.Queue("prices").
	Group("repricer").
	SubscriptionMode(queen.SubscriptionModeAll).
	Conflation(true). // only the newest message of each partition
	Wait(false).
	Pop(ctx)
if err != nil {
	log.Fatal(err)
}
latest := newest[0]
fmt.Println(latest.Data) // map[price:12 sku:A-1]: 10 and 11 are skipped
_, err = client.Ack(ctx, latest, true, queen.AckOptions{ConsumerGroup: "repricer"})
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
let newest = queen
    .queue("prices")
    .group("repricer")
    .subscription_mode(SubscriptionMode::All)
    .conflation(true)                       // only the newest message of each partition
    .wait(false)
    .pop()
    .await?;
let latest = &newest[0];
println!("{}", latest.data); // {"price":12,"sku":"A-1"}: 10 and 11 are skipped
queen.ack(latest).await?;
```

PHP:

```php
[$latest] = $queen->queue('prices')
    ->group('repricer')
    ->subscriptionMode('all')
    ->conflation()                       // only the newest message of each partition
    ->wait(false)
    ->pop();
print_r($latest['data']);                // [sku => A-1, price => 12]: 10 and 11 are skipped
$queen->ack($latest, 'completed', ['group' => 'repricer']);
```

C++:

```cpp
json latest = client.queue("prices")
    .group("repricer")
    .subscription_mode("all")
    .conflation()                    // only the newest message of each partition
    .wait(false)
    .pop()
    .at(0);
std::cout << latest["data"] << "\n"; // {"price":12,"sku":"A-1"}: 10 and 11 are skipped
client.ack(latest, true, {{"group", "repricer"}});
```

**Why:** some queues carry commands where only the latest one matters: "recompute the price of A-1". A worker that is behind should do the work once, with the newest input, instead of three times with inputs that are already old.

Conflation is decided by the group's first pop, needs a consumer group, and cannot be combined with `commitOnDelivery`.

## Faster, with no ack: commitOnDelivery

A pop with `.commitOnDelivery()` moves the group's cursor as the messages are handed out. There is no lease and nothing to ack.

JavaScript:

```js
const readings = await queen.queue('metrics')
  .group('live-chart')
  .subscriptionMode('all')
  .commitOnDelivery()                 // the cursor moves as they are handed out: no ack
  .batch(100)
  .wait(false)
  .pop()
```

Python:

```python
readings = await (
    queen.queue("metrics")
    .group("live-chart")
    .subscription_mode("all")
    .commit_on_delivery()             # the cursor moves as they are handed out: no ack
    .batch(100)
    .wait(False)
    .pop()
)
```

Go:

```go
readings, err := client.Queue("metrics").
	Group("live-chart").
	SubscriptionMode(queen.SubscriptionModeAll).
	CommitOnDelivery(true). // the cursor moves as they are handed out: no ack
	Batch(100).
	Wait(false).
	Pop(ctx)
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
let readings = queen
    .queue("metrics")
    .group("live-chart")
    .subscription_mode(SubscriptionMode::All)
    .commit_on_delivery(true) // the cursor moves as they are handed out: no ack
    .batch(100)
    .wait(false)
    .pop()
    .await?;
```

PHP:

```php
$readings = $queen->queue('metrics')
    ->group('live-chart')
    ->subscriptionMode('all')
    ->commitOnDelivery()                 // the cursor moves as they are handed out: no ack
    ->batch(100)
    ->wait(false)
    ->pop();
```

C++:

```cpp
json readings = client.queue("metrics")
    .group("live-chart")
    .subscription_mode("all")
    .commit_on_delivery()  // the cursor moves as they are handed out: no ack
    .batch(100)
    .wait(false)
    .pop();
```

**Why:** a live dashboard or a metrics chart can lose a reading now and then, and would rather not pay for leases and acks. This is at-most-once delivery: a worker that crashes after the pop loses that batch. It works with `pop()` only; `consume()` refuses it.

## Running a consumer for real

JavaScript:

```js
const stop = new AbortController()
process.once('SIGTERM', () => stop.abort())     // a deploy stops the worker cleanly

await queen.queue('tasks')
  .group('workers')
  .subscriptionMode('all')
  .concurrency(4)                               // four handlers at once, on four partitions
  .renewLease(true, 10000)                      // long tasks keep their lease alive
  .idleMillis(60000)                            // or stop after a minute with nothing to do
  .each()
  .consume(async (task) => runTask(task.data), { signal: stop.signal })
```

Python:

```python
stop = asyncio.Event()
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGTERM, stop.set)   # a deploy stops the worker cleanly

async def handle(task):
    await run_task(task["data"])

await (
    queen.queue("tasks")
    .group("workers")
    .subscription_mode("all")
    .concurrency(4)                   # four handlers at once, on four partitions
    .renew_lease(True, 10000)         # long tasks keep their lease alive
    .idle_millis(60000)               # or stop after a minute with nothing to do
    .each()
    .consume(handle, signal=stop)
)
```

Go:

```go
ctx, stop := signal.NotifyContext(ctx, syscall.SIGTERM) // a deploy stops it cleanly
defer stop()

err := client.Queue("tasks").
	Group("workers").
	SubscriptionMode(queen.SubscriptionModeAll).
	Concurrency(4).          // four handlers at once, on four partitions
	RenewLease(true, 10000). // long tasks keep their lease alive
	IdleMillis(60000).       // or stop after a minute with nothing to do
	Each().
	Consume(ctx, func(ctx context.Context, task *queen.Message) error {
		return runTask(task.Data)
	}).
	Execute(ctx) // returns once ctx is cancelled
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
use queen_mq::Cancel;
use tokio::signal::unix::{signal, SignalKind};

let stop = Cancel::new();
let on_sigterm = stop.clone();
tokio::spawn(async move {                   // a deploy stops the worker cleanly
    if let Ok(mut sigterm) = signal(SignalKind::terminate()) {
        sigterm.recv().await;
        on_sigterm.cancel();
    }
});

queen
    .queue("tasks")
    .group("workers")
    .subscription_mode(SubscriptionMode::All)
    .concurrency(4)                         // four handlers at once, on four partitions
    .renew_lease(Duration::from_secs(10))   // long tasks keep their lease alive
    .idle(Duration::from_secs(60))          // or stop after a minute with nothing to do
    .poll_timeout(Duration::from_secs(5))   // a stop waits for the poll in flight (5 s)
    .cancel(stop)
    .consume(|task: Message| async move { run_task(&task.data).await })
    .await?;
```

PHP:

```php
// A deploy's SIGTERM stops the loop cleanly: consume() handles it itself (ext-pcntl).
$queen->queue('tasks')
    ->group('workers')
    ->subscriptionMode('all')
    ->concurrency(4)                     // four pops at once; their handlers run in turn
    ->renewLease(true, 10000)            // leases renewed every 10 s, between tasks
    ->idleMillis(60000)                  // or stop after a minute with nothing to do
    ->each()
    ->consume(fn (array $task) => runTask($task['data']))
    ->execute();
```

C++:

```cpp
static std::atomic<bool> stop{false};            // static: a signal handler sees it
std::signal(SIGTERM, [](int) { stop = true; });  // a deploy stops the worker cleanly

client.queue("tasks")
    .group("workers")
    .subscription_mode("all")
    .concurrency(4)              // four handlers at once, on four partitions
    .renew_lease(true, 10000)    // long tasks keep their lease alive
    .idle_millis(60000)          // or stop after a minute with nothing to do
    .wait(false)                 // short polls: stop and idle are read between them
    .each()
    .consume([](const json& task) { run_task(task["data"]); }, &stop);
```

- `concurrency(n)` runs `n` handlers at once, each on its own partitions.
- `limit(n)` stops after `n` messages, and `idleMillis(ms)` stops after that long with nothing to do.
- `wait(true)` (the default) waits for messages instead of returning empty, up to `timeoutMillis(ms)` (30 seconds by default, at most 60).
- A `signal` stops the loop at once. Messages it had taken but not started go back with a `retry` ack, which spends no retry, so a deploy does not leave partitions waiting for their leases to run out.

**In the other SDKs:** in 2.6.0 only JavaScript gives the unstarted messages back. Python, Go, Rust, PHP and C++ stop after the message they are handling, and the rest wait for their lease. Python, Rust and C++ also notice the stop only between two long polls, so their code sets a short poll timeout. PHP stops on SIGTERM by itself.

**Why:** a worker runs for weeks, through deploys and quiet nights. These options decide how much it does at once, when it stops, and how it gives its work back when it is told to.

## Read it again

Consuming does not delete: the queue is a log, kept for its retention. A new group can start from a point in time and read the same orders again, while the groups that already read them stay where they are.

JavaScript:

```js
await queen.queue('orders')
  .group('audit')                           // a new group, with its own cursor
  .subscriptionFrom('2026-10-22T00:00:00Z') // everything pushed from this time on
  .each()
  .limit(1)
  .consume(async (msg) => {
    console.log('again:', msg.data)
  })
```

Python:

```python
async def audit(msg):
    print("again:", msg["data"])

await (
    queen.queue("orders")
    .group("audit")                              # a new group, with its own cursor
    .subscription_from("2026-10-22T00:00:00Z")   # everything pushed from this time on
    .each()
    .limit(1)
    .consume(audit)
)
```

Go:

```go
err = client.Queue("orders").
	Group("audit").                           // a new group, with its own cursor
	SubscriptionFrom("2026-10-22T00:00:00Z"). // everything pushed from this time on
	Each().
	Limit(1).
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		fmt.Println("again:", msg.Data)
		return nil
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
queen
    .queue("orders")
    .group("audit")                            // a new group, with its own cursor
    .subscription_from("2026-10-22T00:00:00Z") // everything pushed from this time on
    .limit(1)
    .consume(move |msg: Message| {
        async move {
            println!("again: {}", msg.data);
            Ok::<_, queen_mq::Error>(())
        }
    })
    .await?;
```

PHP:

```php
$queen->queue('orders')
    ->group('audit')                           // a new group, with its own cursor
    ->subscriptionFrom('2026-10-22T00:00:00Z') // everything pushed from this time on
    ->each()
    ->limit(1)
    ->consume(function (array $msg) {
        print_r($msg['data']);
    })
    ->execute();
```

C++:

```cpp
client.queue("orders")
    .group("audit")                              // a new group, with its own cursor
    .subscription_from("2026-10-22T00:00:00Z")   // everything pushed from this time on
    .each()
    .limit(1)
    .consume([&](const json& msg) {
        std::cout << "again: " << msg["data"] << std::endl;
    });
```

Or read a partition by offset with `fetch`: no group, no lease, no ack. It suits an audit, rebuilding one entity, or a sink that keeps its own position.

JavaScript:

```js
const read = await queen.queue('orders').fetch(['customer-42']) // no group, lease or ack
for (const record of read.get('customer-42').records) {
  console.log(record.offset, record.payload)
}
```

Python:

```python
read = await queen.queue("orders").fetch(["customer-42"])  # no group, lease or ack
for record in read["customer-42"]["records"]:
    print(record["offset"], record["payload"])
```

Go:

```go
read, err := client.Queue("orders").Fetch(ctx, []string{"customer-42"}) // no group, lease or ack
if err != nil {
	log.Fatal(err)
}
for _, record := range read["customer-42"].Records {
	fmt.Println(record.Offset, string(record.Payload))
}
```

Rust:

```rust
let read = queen.queue("orders").fetch(["customer-42"]).await?; // no group, lease or ack
for record in &read["customer-42"].records {
    println!("{} {}", record.offset, record.payload);
}
```

PHP:

```php
$read = $queen->queue('orders')->fetch(['customer-42']); // no group, lease or ack
foreach ($read['customer-42']['records'] as $record) {
    echo $record['offset'], ' ', json_encode($record['payload']), "\n";
}
```

C++:

```cpp
auto read = client.queue("orders").fetch({"customer-42"});  // no group, lease or ack
for (const queen::FetchedRecord& record : read["customer-42"].records) {
    std::cout << record.offset << " " << record.payload << std::endl;
}
```

**Why:** history is useful long after the first reader is done with it: to build a new view, to audit what happened, or to rebuild one customer's state from its events.

## See also

- [Acknowledge](/writing-applications/ack/): the four outcomes of an ack.
- [The consumption engine](/architecture/consumption-engine/): how the leader serves pops and leases from memory.
- [HTTP API](/reference/http/): the pop and ack endpoints and every parameter.

Source: https://queenmq.com/writing-applications/pop/index.mdx