Publishing Messages
1. Publishing Basic Message
| Property | Description | Example |
|---|---|---|
| exchange | Name of the exchange to publish to | amq.topic |
| routingKey | Routing key for message routing | logs.info |
| messageBody | Content of the message (bytes or string) | "Hello World" |
| properties | Optional message properties (headers, delivery mode, etc.) | { 'delivery_mode': 2 } |
Example: Publishing a Basic Message
const amqp = require('amqplib');
async function publishMessage() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
const exchange = 'logs';
const routingKey = 'logs.info';
const message = 'Hello, RabbitMQ!';
channel.publish(exchange, routingKey, Buffer.from(message), {
deliveryMode: 2 // persistent
});
console.log('Message published');
await channel.close();
await connection.close();
}
publishMessage().catch(console.error);
2. Setting Routing Key
| Property | Description | Values |
|---|---|---|
| routing_key | Used to route messages to queues based on the exchange type | string |
| exchange_type | Type of exchange that determines how messages are routed | direct, fanout, topic, headers |
Example: Publishing with routing key
const amqp = require('amqplib');
async function publishWithRoutingKey() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
const exchangeName = 'logs';
const routingKey = 'error';
const message = 'An error occurred';
await channel.assertExchange(exchangeName, 'direct', { durable: true });
channel.publish(exchangeName, routingKey, Buffer.from(message));
await channel.close();
await connection.close();
}
publishWithRoutingKey();
Note: The
routing_key is only used by direct and topic exchanges. For fanout exchanges, all messages are delivered to all queues bound to the exchange.
3. Publishing to Default Exchange
| Property | Value | Description |
|---|---|---|
| exchange | "" |
Default exchange name (empty string) |
| routingKey | message.routing.key |
Routing key used to determine queue binding |
| messageBody | "Hello, RabbitMQ!" |
Content of the message to publish |
Example: Publishing to Default Exchange
const amqp = require('amqplib');
async function publishToDefaultExchange() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
const exchange = '';
const routingKey = 'my.routing.key';
const message = 'Hello, RabbitMQ!';
channel.publish(exchange, routingKey, Buffer.from(message));
console.log('Message published to default exchange');
await channel.close();
await connection.close();
}
publishToDefaultExchange().catch(console.error);
Note: When publishing to the default exchange (empty string), messages are routed based on the routing key matching queue names directly.
4. Setting Mandatory Flag
When publishing messages to RabbitMQ, the mandatory flag controls what happens if a message cannot be routed to any queue.
| Flag | Description | Behavior |
|---|---|---|
| mandatory | Boolean flag indicating whether the publisher wants to be notified if a message cannot be routed | If true, the broker will return the message with a basic.nack if no queue is bound to the routing key. If false, the message is silently dropped if no queue matches. |
Example: Publishing with Mandatory Flag
const amqp = require('amqplib');
async function publishWithMandatory() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
const exchangeName = 'myExchange';
const routingKey = 'myRoutingKey';
// Publish with mandatory flag set to true
channel.publish(exchangeName, routingKey, Buffer.from('Hello World'), {
mandatory: true
});
// Handle returned messages (only if mandatory is true and no queue matches)
channel.on('return', (msg) => {
console.log('Message returned:', msg);
});
await channel.close();
await connection.close();
}
publishWithMandatory().catch(console.error);
Note: When
mandatory is set to true, the publisher must handle returned messages to avoid losing data. If no queue is bound to the routing key, the message will be returned and must be handled accordingly.
5. Handling Unroutable Messages
When a message cannot be routed to any queue, it becomes unroutable. RabbitMQ provides mechanisms to handle such messages.
| Property | Description | Default Value |
|---|---|---|
| mandatory | When set to true, the broker will return unroutable messages to the publisher. | false |
| immediate | When set to true, the broker will return unroutable messages immediately if no queue is ready to accept them. | false |
Example: Publishing with Mandatory Flag
const amqp = require('amqplib');
async function publishWithMandatory() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
const exchangeName = 'myExchange';
const routingKey = 'nonexistent.routing.key';
// Publish with mandatory flag set to true
channel.publish(exchangeName, routingKey, Buffer.from('Hello World'), {
mandatory: true
}, (err, ok) => {
if (err) {
console.error('Message returned:', err);
} else {
console.log('Message published successfully');
}
});
channel.close();
connection.close();
}
publishWithMandatory();
Note: When
mandatory is set to true, the broker will return unroutable messages via a basic.return method. Publishers must handle these returns to avoid losing messages.
Warning: Using
immediate is deprecated and discouraged in modern RabbitMQ versions. It can cause performance issues and is not supported by all AMQP 0-9-1 implementations.
6. Using Publisher Confirms
Publisher confirms ensure that messages are successfully delivered to the broker. They provide a way to acknowledge message delivery and handle potential failures.
| Property | Description | Default Value | Notes |
|---|---|---|---|
| publisher_confirms | Enables publisher confirms for a channel | false | Must be set before publishing messages |
| confirm.select | AMQP command to enable publisher confirms | Used in protocol-level configuration |
Example: Enabling Publisher Confirms
const amqp = require('amqplib');
async function enablePublisherConfirms() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
// Enable publisher confirms
await channel.confirmSelect();
console.log('Publisher confirms enabled');
}
enablePublisherConfirms().catch(console.error);
Example: Publishing with Confirm Callbacks
const amqp = require('amqplib');
async function publishWithConfirms() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
await channel.confirmSelect();
// Publish a message with confirm callback
channel.publish('my_exchange', 'my_routing_key', Buffer.from('Hello World'), {
confirm: true
}, (err, ok) => {
if (err) {
console.error('Message failed to publish:', err);
} else {
console.log('Message published successfully');
}
});
}
publishWithConfirms().catch(console.error);
Note: Publisher confirms add overhead to message publishing, so they should be used judiciously in high-throughput scenarios.
7. Batching Messages
Example: Publishing Batches of Messages
const amqp = require('amqplib');
async function publishBatch() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
const queue = 'batch-queue';
await channel.assertQueue(queue, { durable: true });
const messages = [
{ id: 1, data: 'Message 1' },
{ id: 2, data: 'Message 2' },
{ id: 3, data: 'Message 3' }
];
messages.forEach(msg => {
channel.sendToQueue(queue, Buffer.from(JSON.stringify(msg)));
});
await channel.close();
await connection.close();
}
publishBatch().catch(console.error);
| Property | Description | Values |
|---|---|---|
| batchSize | Maximum number of messages to batch before publishing | 100 |
| batchTimeout | Time in milliseconds to wait before publishing a batch | 1000 |
| batchMode | Batching strategy: time-based or size-based |
size-based |
Note: Batching can improve throughput but may increase latency.
Warning: Ensure that batched messages are handled correctly in case of failures.
8. Setting Message Priority
| Property | Value | Description |
|---|---|---|
| priority | 0-9 |
Message priority level (0 = lowest, 9 = highest) |
Example: Setting Priority in AMQP
const amqp = require('amqplib');
async function publishWithPriority() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
const message = 'High priority message';
const priority = 9;
channel.publish(
'my_exchange',
'my_routing_key',
Buffer.from(message),
{ priority: priority }
);
await channel.close();
await connection.close();
}
publishWithPriority();
Note: Message priority is only effective when using a queue that supports prioritized messages (e.g.,
priority argument in queue declaration).
9. Publishing Large Messages
| Property | Value | Description |
|---|---|---|
| message.size | 1MB+ |
Maximum size of a single message |
| max.message.bytes | 10485760 |
Maximum message size in bytes (10MB default) |
| chunking.enabled | false |
Whether message chunking is supported |
Example: Publishing Large Message
const amqp = require('amqplib');
const fs = require('fs');
async function publishLargeMessage() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
const largePayload = fs.readFileSync('large-file.bin');
const exchangeName = 'large-messages';
const routingKey = 'data';
await channel.assertExchange(exchangeName, 'direct', { durable: true });
await channel.publish(exchangeName, routingKey, largePayload, {
headers: { 'content-type': 'application/octet-stream' },
persistent: true
});
console.log('Large message published');
await channel.close();
await connection.close();
}
publishLargeMessage().catch(console.error);
Note: Large messages may impact broker performance and memory usage. Consider using message chunking or file-based approaches for very large payloads.
10. Using Transactional Publishes
Transactional publishes ensure that messages are published atomically within a transaction. This is useful when you need to publish multiple messages or combine publishing with other operations in a single atomic unit.
| Method | Description | Notes |
|---|---|---|
| txSelect() | Enables transaction mode for the channel. | Must be called before any transactional operations. |
| txCommit() | Commits the current transaction. | Any messages published since txSelect() are now made visible. |
| txRollback() | Rolls back the current transaction. | All messages published since txSelect() are discarded. |
Example: Publishing with Transactional Semantics
const amqp = require('amqplib');
async function publishWithTransaction() {
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
try {
// Enable transaction mode
await channel.txSelect();
// Publish multiple messages
await channel.publish('my_exchange', 'routing.key', Buffer.from('Message 1'));
await channel.publish('my_exchange', 'routing.key', Buffer.from('Message 2'));
// Commit the transaction
await channel.txCommit();
console.log('All messages published successfully');
} catch (err) {
// Rollback on error
await channel.txRollback();
console.error('Transaction failed, rolled back', err);
} finally {
await channel.close();
await connection.close();
}
}
Note: Transactional publishes can impact performance due to the overhead of managing transaction state. Use them only when atomicity is required.
11. Measuring Publish Rate
| Property | Value | Description |
|---|---|---|
| publish.rate | messages/second |
Number of messages published per second |
| publish.bytes | bytes/second |
Amount of data published per second |
| message.size.avg | bytes |
Average size of published messages |
Example: Monitoring publish rate via RabbitMQ Management Plugin
curl -u guest:guest http://localhost:15672/api/queues/%2f/my-queue
Note: The publish rate can be monitored using the
rabbitmq_management plugin or via Prometheus metrics if configured.
12. Handling Publish Errors
| Error Type | Description | Handling Strategy |
|---|---|---|
| ChannelClosedError | Occurs when the channel is closed unexpectedly during publish. | Reconnect and redeclare queues/exchanges before retrying publish. |
| MessageTooLargeError | Published message exceeds broker's max frame size. | Reduce message size or split into smaller chunks. |
| AccessRefusedError | Publisher lacks permissions to publish to a specific exchange. | Verify user permissions and ACLs on the exchange. |
| NotFoundError | Exchange or queue does not exist at time of publish. | Ensure exchange/queue is declared before publishing. |
Example: Error Handling in Publisher
const amqp = require('amqplib');
async function publishWithRetry() {
const conn = await amqp.connect('amqp://localhost');
const ch = await conn.createChannel();
try {
await ch.publish('my-exchange', 'routing-key', Buffer.from('Hello World'));
} catch (err) {
if (err.name === 'ChannelClosedError') {
console.error('Channel closed, reconnecting...');
// Reconnect logic here
} else if (err.name === 'MessageTooLargeError') {
console.error('Message too large, reducing size...');
// Reduce message size logic here
} else {
console.error('Publish error:', err);
}
}
await ch.close();
await conn.close();
}
Note: Always implement retry logic with exponential backoff for transient errors like
ChannelClosedError.