Optimizing Performance

const amqp = require('amqplib');

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

  // Set prefetch count to 10
  channel.prefetch(10);

  // Now consumer will receive max 10 unacknowledged messages
  channel.consume('queue-name', (msg) => {
    console.log("Received:", msg.content.toString());
    channel.ack(msg);
  });
}

const amqp = require('amqplib');

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

  const queueName = 'batch-queue';
  await channel.assertQueue(queueName, { durable: true });

  const messages = [
    { id: 1, content: 'Message 1' },
    { id: 2, content: 'Message 2' },
    { id: 3, content: 'Message 3' }
  ];

  // Publish messages in a batch
  for (const msg of messages) {
    channel.publish('', queueName, Buffer.from(JSON.stringify(msg)), {
      persistent: true
    });
  }

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

publishBatch().catch(console.error);

const amqp = require('amqplib');

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

  const queueName = 'batch-queue';
  await channel.assertQueue(queueName, { durable: true });

  // Set up consumer to process messages in batches
  channel.consume(queueName, async (msg) => {
    if (msg) {
      console.log('Received:', msg.content.toString());
      // Acknowledge the message in a batch
      channel.ack(msg);
    }
  }, { noAck: false });

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

acknowledgeBatch().catch(console.error);

import com.rabbitmq.client.ConnectionFactory;

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
factory.setConnectionTimeout(30000);
factory.setRequestedHeartbeat(60);

// Set pool size
factory.setConnectionTimeout(30000);

listeners.tcp.default = 5672
max_message_size = 1048576
frame_max = 131072

const queueOptions = {
  durable: true,
  autoDelete: false,
  arguments: {
    'x-queue-type': 'lazy'
  }
};

channel.assertQueue('myLazyQueue', queueOptions);

const queueOptions = {
  durable: true,
  autoDelete: false,
  arguments: {
    'x-max-length': 1000,
    'x-overflow': 'drop-head'
  }
};

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

const msg = {
  content: Buffer.from("Hello World"),
  properties: {
    delivery_mode: 2 // persistent
  }
};

disk_free_limit: 100MB

# Increase max ports to handle more concurrent connections
vm_memory_high_watermark = 0.4
erlang.max_ports = 2048

# Increase max ETS tables for large routing tables
erlang.max_ets_tables = 512

# Set number of schedulers to match CPU cores
erlang.schedulers = 8

const amqp = require('amqplib');

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

  // Enable flow control
  await channel.flow(true);

  // Set prefetch count to limit messages in flight
  await channel.prefetch(10);

  console.log('Flow control enabled');
}

enableFlowControl().catch(console.error);

rabbitmqctl status | grep mem_used

scrape_configs:
  - job_name: 'rabbitmq'
    static_configs:
      - targets: ['localhost:15692']

rabbitmq-perf-test -a 500 -c 5 -p 50