Configuring Quality of Service
1. Setting Prefetch Count
| Property | Value | Description |
|---|---|---|
| prefetchCount | 10 |
Number of messages to prefetch from the broker |
Example: Configuring Prefetch Count in JavaScript
const amqp = require('amqplib');
async function configurePrefetch() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set prefetch count to 10
await channel.prefetch(10);
console.log('Prefetch count set to 10');
}
configurePrefetch().catch(console.error);
2. Setting Prefetch Size
| Property | Values | Description |
|---|---|---|
| prefetchSize | 0 |
Maximum size of prefetch in bytes. Default is 0 (no limit). |
Example: Setting Prefetch Size
const amqp = require('amqplib');
async function connect() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set prefetch size to 1024 bytes (1KB)
await channel.prefetch(0, 1024);
console.log('Prefetch size set to 1KB');
}
connect().catch(console.error);
3. Using Channel-Level QoS
| Property | Value | Description |
|---|---|---|
| qos.prefetchCount | 0 |
Number of messages to prefetch per channel |
| qos.global | false |
Whether QoS applies to entire connection or individual channels |
Example: Setting Channel-Level QoS
const amqp = require('amqplib');
async function setupQoS() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set QoS for this channel
await channel.prefetch(10); // Prefetch up to 10 messages
console.log('Channel QoS configured');
}
setupQoS().catch(console.error);
Note: Setting
prefetch(0) disables prefetching, meaning all messages are delivered immediately.
Warning: QoS settings are applied per channel, not per consumer. If multiple consumers use the same channel, they will share the same QoS limits.
4. Using Global QoS
| Property | Values | Description |
|---|---|---|
| global | true / false |
When set to true, QoS settings apply to all consumers on the channel. When false, QoS is applied per consumer. |
| prefetchCount | 0 or positive integer |
Maximum number of unacknowledged messages that can be delivered to a consumer. Setting to 0 disables the limit. |
| prefetchSize | 0 or positive integer |
Maximum total size of unacknowledged messages in bytes. Setting to 0 disables the limit. |
Example: Setting Global QoS
const amqp = require('amqplib');
async function setupGlobalQoS() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set global QoS to limit 10 messages per consumer
await channel.qos(10, 0, true);
console.log('Global QoS applied with prefetchCount=10');
}
setupGlobalQoS().catch(console.error);
Note: Global QoS settings apply to all consumers on the channel and are effective immediately after setting.
5. Configuring Consumer Prefetch
| Property | Value | Description |
|---|---|---|
| qos.prefetchCount | 10 |
Number of messages to prefetch per consumer |
| qos.global | false |
Whether prefetch setting applies to all consumers on channel |
Example: Setting Prefetch Count
const amqp = require('amqplib');
async function configureQoS() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set prefetch count to 5 messages
await channel.prefetch(5);
console.log('QoS prefetch set to 5');
}
configureQoS().catch(console.error);
Note: Prefetch count is applied per consumer, not globally across all consumers.
Warning: Setting prefetch count too high can lead to memory issues if consumers are slow to process messages.
6. Understanding QoS Impact
Note: Quality of Service (QoS) settings in RabbitMQ control how messages are delivered and processed, affecting system throughput, latency, and message ordering.
| Setting | Description | Default Value | Impact |
|---|---|---|---|
| prefetch_count | Maximum number of unacknowledged messages a consumer can receive | 0 |
Controls consumer throughput and prevents overloading |
| prefetch_size | Maximum total size of unacknowledged messages (in bytes) | 0 |
Helps manage memory usage for consumers with large messages |
| global | Applies QoS settings to all consumers on the channel, not just the current one | false |
When true, affects all consumers in the channel; when false, applies per-consumer |
Example: Setting QoS for a Consumer
const amqp = require('amqplib');
async function setupQoS() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set QoS to limit to 10 unacknowledged messages
channel.prefetch(10);
// Now consume messages with QoS applied
await channel.consume('my-queue', (msg) => {
console.log("Received:", msg.content.toString());
channel.ack(msg);
});
}
setupQoS().catch(console.error);
QoS Impact Summary
- Throughput Control: Limits how many messages a consumer can process at once, preventing resource exhaustion.
- Memory Management:
prefetch_sizeensures consumers don't receive messages that exceed memory limits. - Fair Distribution: When
globalis false, QoS is applied per-consumer, ensuring fair load distribution.
7. Tuning Prefetch for Performance
| Parameter | Description | Default | Notes |
|---|---|---|---|
| prefetch_count | Maximum number of unacknowledged messages a consumer can prefetch | 0 (no limit) | Set to 1 for fair dispatch, or higher for performance |
| prefetch_size | Maximum total size of unacknowledged messages in bytes | 0 (no limit) | Useful for memory-constrained consumers |
| global | Whether prefetch settings apply to all consumers on the channel | false | When true, settings apply to all consumers in the channel |
Example: Setting Prefetch Count
const amqp = require('amqplib');
async function setupConsumer() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set prefetch count to 10
channel.prefetch(10);
await channel.consume('my-queue', (msg) => {
console.log("Processing message", msg.content.toString());
channel.ack(msg);
});
}
Note: Setting
prefetch_count to 1 ensures fair distribution of messages among consumers, preventing one consumer from overwhelming others.
Warning: Setting
prefetch_size too low may cause frequent message delivery, while setting it too high can lead to memory pressure.
8. Using QoS with Auto-Ack
| Property | Description | Default Value | Notes |
|---|---|---|---|
| qos.prefetchCount | Number of messages to prefetch | 0 (unlimited) | Set to 1 for fair dispatch |
| qos.global | Apply QoS globally or per-consumer | false | When true, applies to all consumers on channel |
Example: Setting QoS with Auto-Ack
const amqp = require('amqplib');
async function setupQoS() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set QoS to prefetch 1 message at a time
channel.prefetch(1);
// Consume messages with auto-ack enabled
channel.consume('my-queue', (msg) => {
console.log("Received:", msg.content.toString());
// Message is automatically acknowledged
}, { noAck: true });
}
setupQoS().catch(console.error);
Note: With
noAck: true, the broker acknowledges messages immediately upon delivery, so QoS prefetch count limits how many unacknowledged messages are in flight at once.
9. Combining QoS with Priority
| Property | Description | Values | Notes |
|---|---|---|---|
| qos.prefetchCount | Number of messages to prefetch | 10 |
Applies to all consumers on the channel |
| qos.global | Whether QoS applies globally or per-consumer | false |
Defaults to false, applies per-consumer |
Example: Setting QoS with Priority
const amqp = require('amqplib');
async function setupQosWithPriority() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set QoS to prefetch 5 messages
await channel.prefetch(5);
// Set priority for a specific queue
const queueName = 'priority-queue';
await channel.assertQueue(queueName, {
durable: true,
arguments: {
'x-max-priority': 10
}
});
// Consume messages with priority handling
await channel.consume(queueName, (msg) => {
const priority = msg.properties.priority || 0;
console.log(`Processing message with priority ${priority}`);
channel.ack(msg);
}, { noAck: false });
}
setupQosWithPriority().catch(console.error);
Note: QoS settings apply per channel, not per consumer. When
global is set to true, QoS settings are applied across all consumers on the channel.
Warning: If
prefetchCount is set too high, it can lead to memory pressure and slow processing. Adjust based on consumer capacity.
10. Monitoring Prefetch Metrics
| Metric | Description | Unit | Notes |
|---|---|---|---|
| prefetch.count | Number of messages currently in flight | messages | Indicates how many unacknowledged messages are being processed by consumers |
| prefetch.limit | Maximum number of messages that can be in flight | messages | Configured value for prefetch count, set via channel.basicQos() |
| prefetch.global | Whether prefetch is applied globally or per-channel | boolean | True if prefetch settings apply to all channels in the connection |
Example: Setting Prefetch Limit
const amqp = require('amqplib');
async function setupPrefetch() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Set prefetch to 10 messages
await channel.prefetch(10);
console.log('Prefetch limit set to 10');
}
setupPrefetch().catch(console.error);
Note: Prefetch settings are applied per-channel and affect how many messages a consumer can have in flight before acknowledging them.