rabbitmq

---

1. Simple Message Publishing and Consumption

// Publish message
connection.createChannel((err, channel) => {
  channel.assertQueue('my-queue');
  channel.sendToQueue('my-queue', Buffer.from('Hello World!'));
});

// Consume message
connection.createChannel((err, channel) => {
  channel.assertQueue('my-queue');
  channel.consume('my-queue', (msg) => {
    console.log(msg.content.toString());
  });
});

2. Message Prioritization

// Publish message with priority
channel.sendToQueue('my-queue', Buffer.from('High Priority Message'), {
  priority: 10,
});

// Consume message by priority
channel.consume('my-queue', (msg) => {
  console.log(msg.content.toString(), `Priority: ${msg.properties.priority}`);
}, {
  noAck: true,
});

3. Message Expiration

// Publish message with expiration
channel.sendToQueue('my-queue', Buffer.from('Message Expires in 5s'), {
  expiration: 5000,
});

// Consume message before expiration
channel.consume('my-queue', (msg) => {
  console.log(msg.content.toString());
});

4. Dead Letter Exchange

// Declare dead letter exchange
channel.assertExchange('my-dead-letter-exchange', 'direct');
channel.bindQueue('my-dead-letter-queue', 'my-dead-letter-exchange', 'my-dl-routing-key');

// Declare original exchange and bind to dead letter exchange
channel.assertExchange('my-exchange', 'topic');
channel.bindExchange('my-exchange', 'my-dead-letter-exchange', 'my-dl-routing-key');

// Publish message to original exchange
channel.publish('my-exchange', 'my-routing-key', Buffer.from('Message to Dead Letter Exchange'));

5. Message Retries

// Define retry policy with max 5 attempts and delay between attempts
channel.publish('my-exchange', 'my-routing-key', Buffer.from('Message with Retries'), {
  retry: {
    max: 5,
    delay: 100,
  },
});

6. Message Deduplication

// Define deduplication strategy
channel.prefetch(1);
channel.consume('my-queue', (msg) => {
  channel.nack(msg, false, true);
});

7. Publish Confirms

// Enable publish confirms
channel.confirmChannel();

// Publish message
channel.sendToQueue('my-queue', Buffer.from('Message with Publish Confirm'));

8. Transactional Publish

// Start transaction
channel.tx(async (err, tx) => {
  if (err) throw err;

  try {
    // Publish message in transaction
    await tx.publish('my-exchange', 'my-routing-key', Buffer.from('Message in Transaction'));

    // Commit transaction
    await tx.commit();
  } catch (err) {
    // Rollback transaction on error
    await tx.rollback();
  }
});

9. Consumer Prefetching

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

// Consume messages
channel.consume('my-queue', (msg) => {
  // Process message
});

10. Consumer Acknowledgements

// Consume message and acknowledge it
channel.consume('my-queue', (msg) => {
  channel.ack(msg);
});

11. Consumer Nacks

// Consume message and nack it
channel.consume('my-queue', (msg) => {
  channel.nack(msg);
});

12. Exchange Types (Direct)

// Declare direct exchange
channel.assertExchange('my-direct-exchange', 'direct');

// Bind queue to exchange
channel.bindQueue('my-direct-queue', 'my-direct-exchange', 'my-direct-routing-key');

// Publish message to exchange
channel.publish('my-direct-exchange', 'my-direct-routing-key', Buffer.from('Message to Direct Exchange'));

13. Exchange Types (Fanout)

// Declare fanout exchange
channel.assertExchange('my-fanout-exchange', 'fanout');

// Bind queue to exchange
channel.bindQueue('my-fanout-queue', 'my-fanout-exchange', '');

// Publish message to exchange
channel.publish('my-fanout-exchange', '', Buffer.from('Message to Fanout Exchange'));

14. Exchange Types (Topic)

// Declare topic exchange
channel.assertExchange('my-topic-exchange', 'topic');

// Bind queue to exchange
channel.bindQueue('my-topic-queue', 'my-topic-exchange', 'my-topic-routing-key.*');

// Publish message to exchange
channel.publish('my-topic-exchange', 'my-topic-routing-key.test', Buffer.from('Message to Topic Exchange'));

15. Message Persistence

// Set queue durability
channel.assertQueue('my-durable-queue', {
  durable: true,
});

16. Message Persistence Confirmation

// Enable publisher confirms
channel.confirmChannel();

// Publish message and wait for confirmation
channel.waitForConfirms(true);
channel.sendToQueue('my-durable-queue', Buffer.from('Message with Persistence Confirmation'));

17. Consumer Recovery

// Use channel reconnect strategy
channel.addSetup((ch) => {
  ch.on('error', (err) => {
    if (err.code === 'ECONNRESET') {
      ch.setup();
    }
  });
});

18. Queue Auto-Deletion

// Declare auto-delete queue
channel.assertQueue('my-auto-delete-queue', {
  autoDelete: true,
});

19. Queue Expiration

// Declare queue with expiration
channel.assertQueue('my-expired-queue', {
  expires: 3600000, // 1 hour in milliseconds
});

20. Exclusive Queue

// Declare exclusive queue
channel.assertQueue('my-exclusive-queue', {
  exclusive: true,
});

21. Queue Message Limit

// Declare queue with message limit
channel.assertQueue('my-message-limit-queue', {
  messageLimit: 100,
});

22. Queue Size Limit

// Declare queue with size limit
channel.assertQueue('my-size-limit-queue', {
  maxLength: 102400, // 100 KB
});

23. Prefetch Size Limit

// Set prefetch size limit
channel.prefetch(null, 102400); // 100 KB

24. Message TTL

// Publish message with TTL
channel.sendToQueue('my-ttl-queue', Buffer.from('Message with TTL'), {
  ttl: 30000, // 30 seconds
});

25. RabbitMQ Management Plugin

// Use RabbitMQ Management Plugin
const management = require('rabbit-management');

// Connect to management API
const api = management('amqp://localhost:15672', 'guest', 'guest');

// List all queues
api.listQueues((err, queues) => {
  if (err) throw err;

  queues.forEach((queue) => console.log(queue.name));
});