Understanding Publisher Confirms

1. Enabling Confirm Mode

Confirm mode in RabbitMQ ensures that publishers receive acknowledgments for each message they publish, allowing them to detect and handle delivery failures.

Method	Description	Notes
channel.confirmSelect()	Enables confirm mode for the channel	Must be called before publishing messages to enable confirmation

Example: Enabling Confirm Mode

const amqp = require('amqplib');

async function enableConfirmMode() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Enable confirm mode
  await channel.confirmSelect();
  console.log('Confirm mode enabled');

  // Now all published messages will be confirmed
  channel.publish('my-exchange', 'my-routing-key', Buffer.from('Hello World'), {
    persistent: true
  });

  channel.on('ack', (ok) => {
    console.log('Message confirmed');
  });

  channel.on('nack', (err) => {
    console.error('Message not confirmed:', err);
  });
}

enableConfirmMode().catch(console.error);

Note: Once confirm mode is enabled, all messages published on that channel will be confirmed. It's not possible to selectively confirm specific messages.

2. Handling Basic Ack

Property	Description	Values
basic.ack	Indicates if the message was acknowledged by the broker	`true` or `false`

Example: Handling Basic Ack in Java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;

Channel channel = connection.createChannel();

channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        System.out.println("Message confirmed: " + deliveryTag);
    }

    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        System.out.println("Message not confirmed: " + deliveryTag);
    }
});

3. Handling Basic Nack

Property	Description	Values
basic.nack	Indicates a negative acknowledgment from the broker	`true`

Example: Handling Nack in Publisher Confirms

const amqp = require('amqplib');

async function handleNack() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  channel.on('basic.nack', (deliveryTag, multiple) => {
    console.log(`Nack received for delivery tag: ${deliveryTag}`);
  });

  channel.confirmSelect();
  channel.publish('exchange', 'routing.key', Buffer.from('message'), {
    messageId: 'msg123'
  });
}

handleNack().catch(console.error);

Note: When a basic.nack is received, it indicates that the message was not successfully delivered to a queue or processed by a consumer. Publishers should implement retry logic or error handling based on this event.

4. Using Delivery Tags

Property	Description	Usage
delivery_tag	Unique identifier for a message within a channel	Used to acknowledge or negatively acknowledge messages
multiple	Flag to acknowledge multiple messages at once	Set to `true` to acknowledge all messages up to and including the specified delivery tag

Example: Acknowledging Messages with Delivery Tags

const amqp = require('amqplib');

async function processMessages() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  channel.on('confirm', (deliveryTag, multiple) => {
    if (multiple) {
      console.log(`Acknowledged up to delivery tag ${deliveryTag}`);
    } else {
      console.log(`Acknowledged delivery tag ${deliveryTag}`);
    }
  });

  channel.publish('my-exchange', 'my-routing-key', Buffer.from('Hello World'), {
    messageId: 'msg-123',
    timestamp: Date.now()
  });
}

processMessages().catch(console.error);

Note: Delivery tags are unique within a channel and can be used to track individual messages for confirmation.

5. Implementing Synchronous Confirms

Implementing synchronous confirms in RabbitMQ involves using the confirmSelect method to enable publisher confirms and then waiting for the broker's confirmation before proceeding.

Method	Description	Notes
confirmSelect	Enables publisher confirms on a channel	Must be called before publishing messages
waitForConfirm	Blocks until the next confirm is received	Useful for synchronous confirm handling

Example: Synchronous Confirm Implementation

const amqp = require('amqplib');

async function implementSynchronousConfirms() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Enable publisher confirms
  await channel.confirmSelect();

  try {
    // Publish a message
    const msg = 'Hello, RabbitMQ!';
    await channel.publish('', 'my-queue', Buffer.from(msg));

    // Wait for the confirm
    await channel.waitForConfirm();

    console.log('Message confirmed successfully');
  } catch (err) {
    console.error('Message not confirmed:', err);
  }

  channel.close();
  connection.close();
}

implementSynchronousConfirms().catch(console.error);

Note: Synchronous confirms can significantly impact performance due to blocking operations. Consider using asynchronous confirms for better throughput.

6. Implementing Asynchronous Confirms

Example: Basic Async Confirm Setup

const amqp = require('amqplib');

async function setupAsyncConfirms() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Enable confirms
  channel.addConfirmListener((err, ok) => {
    if (err) {
      console.error('Confirm failed:', err);
    } else {
      console.log('Message confirmed');
    }
  });

  // Publish with async confirm
  channel.publish('exchange', 'routing.key', Buffer.from('Hello World'), {
    confirm: true
  });

  await channel.close();
  await connection.close();
}

setupAsyncConfirms().catch(console.error);

Method	Description	Notes
addConfirmListener	Registers a callback for confirm events	Called once per confirmed message
publish	Publishes a message with confirm option	Must set `confirm: true`

Note: Confirms are processed asynchronously, so order is not guaranteed.

Example: Handling Multiple Confirms

const pendingConfirms = new Map();

channel.addConfirmListener((err, ok) => {
  const deliveryTag = ok.deliveryTag;
  const resolve = pendingConfirms.get(deliveryTag);

  if (resolve) {
    pendingConfirms.delete(deliveryTag);
    resolve();
  }
});

async function publishWithConfirm(exchange, routingKey, message) {
  const confirmPromise = new Promise((resolve) => {
    const deliveryTag = channel.nextPublishSeqNo;
    pendingConfirms.set(deliveryTag, resolve);
  });

  channel.publish(exchange, routingKey, message, {
    confirm: true
  });

  await confirmPromise;
}

7. Batching Confirms

Example: Batching Publisher Confirms

const amqp = require('amqplib');
const batchSize = 10;

async function batchConfirms() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Enable publisher confirms
  await channel.confirmSelect();

  let confirmCount = 0;
  const pendingConfirms = new Set();

  channel.on('confirm', (deliveryTag) => {
    pendingConfirms.delete(deliveryTag);
    confirmCount++;
    
    if (confirmCount % batchSize === 0) {
      console.log(`Processed ${confirmCount} confirms in batch`);
    }
  });

  // Send messages in batches
  for (let i = 0; i < 100; i++) {
    const msg = `Message ${i}`;
    const deliveryTag = channel.publish('exchange', 'routing.key', Buffer.from(msg));
    pendingConfirms.add(deliveryTag);
  }

  // Wait for all confirms to be processed
  while (pendingConfirms.size > 0) {
    await new Promise(resolve => setTimeout(resolve, 10));
  }

  console.log('All confirms processed');
  await channel.close();
  await connection.close();
}

batchConfirms().catch(console.error);

Property	Description	Values
batchSize	Number of messages to batch before processing confirms	`10`
pendingConfirms	Set tracking unconfirmed deliveries	`Set`

Note: Batching confirms can improve performance by reducing the frequency of confirm callbacks.

Warning: Ensure all messages are confirmed before shutting down the channel to avoid message loss.

8. Handling Confirm Timeouts

Property	Description	Default Value	Notes
confirm.timeout	Timeout for publisher confirms in milliseconds	`30000`	Used to detect unconfirmed messages
confirm.retry.count	Number of retries for failed confirms	`3`	Attempts to reconfirm before failing

Example: Setting Confirm Timeout

const amqp = require('amqplib');

async function setupConfirmTimeout() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createConfirmChannel();

  // Set confirm timeout to 5 seconds
  channel.confirmTimeout = 5000;

  // Publish a message
  const msg = 'Hello RabbitMQ';
  channel.publish('my-exchange', 'my-routing-key', Buffer.from(msg), {
    confirm: true
  });

  // Handle timeout
  channel.on('ack', (ok) => {
    console.log('Message confirmed');
  });

  channel.on('nack', (err) => {
    console.error('Message not confirmed:', err);
  });
}

setupConfirmTimeout().catch(console.error);

Note: If a confirm times out, the message is considered lost and should be re-sent or handled according to your application's retry logic.

9. Using Confirms for Reliability

Example: Publisher Confirms Setup

const amqp = require('amqplib');
async function setupConfirms() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  // Enable publisher confirms
  channel.addConfirmListener((err, ok) => {
    if (err) {
      console.error('Publish failed:', err);
    } else {
      console.log('Message confirmed');
    }
  });
  
  return { channel, connection };
}

Property	Description	Values
confirm	Enables publisher confirms on a channel	`true`
messageId	Unique identifier for a message	`string`
deliveryMode	Controls persistence of messages	`1` (transient), `2` (persistent)

Note: Publisher confirms ensure that messages are successfully delivered to the broker and persisted if needed.

Warning: If publisher confirms are not handled properly, messages may be lost or duplicated in case of broker failures.

Reliability Patterns with Publisher Confirms

Idempotent Publishing: Ensure that publishing the same message multiple times does not cause side effects.
Message Acknowledgement: Use confirms to verify that messages are received by the broker.
Retry Logic: Implement retry mechanisms for failed confirmations to maintain reliability.

10. Handling Nack Scenarios

When a publisher receives a nack, it indicates that the broker was unable to process the message. Handling nack scenarios is crucial for ensuring message reliability.

Scenario	Description	Action
Nack with requeue=true	The broker rejects the message and asks it to be requeued for delivery.	Re-publish the message to the same queue or a dead-letter exchange (DLX) if configured.
Nack with requeue=false	The broker rejects the message and does not requeue it.	Implement a dead-letter exchange or move to a separate error queue for manual inspection.
Multiple Nacks	Repeated nack responses may indicate a persistent issue with message handling or broker configuration.	Log the error and consider implementing circuit breaker logic to prevent message flooding.

Example: Handling Nack in Java

import com.rabbitmq.client.*;

public class NackHandler {
    public static void main(String[] args) throws IOException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.confirmSelect(); // Enable publisher confirms

        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) {
                System.out.println("Message confirmed");
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) {
                System.out.println("Message rejected (nack)");
                // Re-publish or move to DLX
            }
        });

        channel.basicPublish("", "queue", null, "Hello".getBytes());
    }
}

Note: Always ensure that your application handles nack scenarios gracefully to avoid message loss or duplication.

Warning: Requeuing messages can lead to infinite loops if not handled carefully. Implement retry limits or DLX routing to prevent this.

11. Combining Confirms with Transactions

Combining publisher confirms with transactions in RabbitMQ allows for reliable message delivery with atomicity guarantees. This combination ensures that either all messages in a transaction are delivered successfully, or none are.

Feature	Description	Notes
Transaction Mode	Enables transactional behavior for publishers	Must be enabled before sending messages
Confirm Mode	Ensures message delivery confirmation	Must be enabled before sending messages
Transaction Commit	Commits all messages in the transaction	Only after confirm mode is enabled

Example: Publishing Messages with Confirms and Transactions

const amqp = require('amqplib');

async function publishWithConfirmsAndTransactions() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Enable confirm mode
  channel.confirmSelect();

  // Start transaction
  channel.transaction();

  // Publish messages
  const message = { text: 'Hello, World!' };
  channel.publish('my-exchange', 'my-routing-key', Buffer.from(JSON.stringify(message)));

  // Commit transaction
  channel.commit();

  // Wait for confirmations
  channel.on('ack', (ok) => console.log('Message confirmed'));
  channel.on('nack', (err) => console.error('Message failed:', err));
}

publishWithConfirmsAndTransactions().catch(console.error);

Note: When using confirms with transactions, the confirm callback will be triggered after the transaction is committed and acknowledged by the broker.

Warning: Transactional publishing can introduce performance overhead due to the need for coordination between publisher and broker.

12. Monitoring Confirm Rate

Property	Description	Values	Notes
confirm.rate	Rate of confirmed messages per second	`float`	Measured in confirms/second
confirm.count	Total number of confirmed messages	`integer`	Accumulative counter
confirm.latency	Average time to confirm a message	`float (ms)`	Latency from publish to confirm

Example: Monitoring Confirm Rate via Management Plugin

curl -u guest:guest http://localhost:15672/api/queues/%2f/my-queue

Example: Using RabbitMQ Metrics in Prometheus

metrics:
  rabbitmq:
    confirm_rate: rabbitmq.confirm.rate
    confirm_count: rabbitmq.confirm.count
    confirm_latency: rabbitmq.confirm.latency

Note: Confirm rate is a key indicator of publisher reliability and system throughput.