Working with Quorum Queues

const amqp = require('amqplib');

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

  const queueName = 'my-quorum-queue';
  const options = {
    durable: true,
    autoDelete: false,
    arguments: { 'x-queue-type': 'quorum' }
  };

  await channel.assertQueue(queueName, options);
  console.log(`Quorum queue '${queueName}' declared successfully`);
  
  await channel.close();
  await connection.close();
}

declareQuorumQueue().catch(console.error);

# Simulated Raft election process
# Node A (leader) fails
# Node B becomes candidate and starts election
# Nodes C and D vote for B
# B becomes new leader and syncs with followers

const amqp = require('amqplib');

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

  const queueName = 'my-quorum-queue';
  const options = {
    durable: true,
    arguments: {
      'x-queue-type': 'quorum',
      'x-quorum-initial-group-size': 5,
      'x-quorum-max-group-size': 7
    }
  };

  await channel.assertQueue(queueName, options);
  console.log(`Quorum queue "${queueName}" created with custom group size.`);
  
  await channel.close();
  await connection.close();
})();

{"name": "my-queue", "type": "quorum", "arguments": {"x-queue-type": "quorum", "x-max-length": 1000}}

{
  "queue": {
    "leader": {
      "failover": true,
      "timeout": "30s",
      "retries": 3
    }
  }
}

const queueOptions = {
  deadLetterExchange: 'dlx.name',
  deadLetterRoutingKey: 'dlx.routing.key',
  messageTtl: 30000
};

// Declare queue with poison handling
await channel.assertQueue('my-queue', queueOptions);

// Declare DLX and queue for poisoned messages
await channel.assertExchange('dlx.name', 'direct');
await channel.assertQueue('poisoned-messages-queue');
await channel.bindQueue('poisoned-messages-queue', 'dlx.name', 'dlx.routing.key');

{
  "name": "queue-with-dlq",
  "arguments": {
    "x-dead-letter-exchange": "dlx.exchange",
    "x-dead-letter-routing-key": "dlq.routing.key"
  }
}

{
  "name": "ttl-dlq-queue",
  "arguments": {
    "x-message-ttl": 30000,
    "x-dead-letter-exchange": "dlx.exchange"
  }
}

{"queue": {"vm_memory_limit": 1073741824}}

const queueOptions = {
  maxLength: 1000,
  overflow: 'reject.publish'
};

// Declare a quorum queue with length limit
await channel.assertQueue('my-quorum-queue', {
  durable: true,
  quorum: true,
  maxLength: queueOptions.maxLength,
  overflow: queueOptions.overflow
});

{queue: "performance-test", type: "quorum", arguments: {x-max-length: 1000000}}

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

rabbitmq:
  queue:
    name: my-quorum-queue
    messages_ready: 10
    consumers: 2

const queueName = 'my-quorum-queue';
const options = {
  durable: true,
  autoDelete: false,
  arguments: {
    'x-queue-type': 'quorum'
  }
};

// Using amqplib to declare a quorum queue
channel.assertQueue(queueName, options);