RabbitMQ 速查表 — 90+ 条命令、模式与常见坑

rabbitmqctl status

rabbitmqctl status --formatter=json

rabbitmqctl list_queues name messages consumers

rabbitmqctl list_queues -p /production name messages consumers unacked_messages memory

rabbitmqctl list_exchanges name type durable

rabbitmqctl list_exchanges -p /myapp name type

rabbitmqctl list_bindings

rabbitmqctl list_bindings -p /myapp source_name destination_name routing_key

rabbitmqctl list_connections name state user

rabbitmqctl list_connections name state user vhost recv_oct send_oct

rabbitmqctl list_channels

rabbitmqctl list_channels name number consumer_count

rabbitmqctl list_consumers

rabbitmqctl list_consumers queue_name consumer_tag ack_required prefetch_count

rabbitmqctl add_user alice secretpassword

rabbitmqctl add_user ci-agent "$(openssl rand -base64 32)"

rabbitmqctl set_user_tags alice administrator

rabbitmqctl set_user_tags ci-agent monitoring

rabbitmqctl set_user_tags alice   # clears all tags

rabbitmqctl set_permissions -p / alice ".*" ".*" ".*"

rabbitmqctl set_permissions -p /prod alice "^app-" "^app-" "^app-"

rabbitmqctl list_users

rabbitmqctl delete_user alice

rabbitmqctl change_password alice newpass123

rabbitmqctl list_permissions -p /

rabbitmqctl list_permissions -p /production

rabbitmqctl add_vhost /production

rabbitmqctl add_vhost /staging

rabbitmqctl list_vhosts

rabbitmqctl list_vhosts name tracing

rabbitmqctl delete_vhost /staging

rabbitmqctl purge_queue payment-queue

rabbitmqctl purge_queue -p /prod orders

rabbitmqctl delete_queue old-test-queue

rabbitmqctl delete_queue -p /prod deadletter

rabbitmqctl delete_exchange old-events

rabbitmqctl delete_exchange -p /prod notifications

rabbitmqctl export_definitions /tmp/rabbitmq-backup.json

rabbitmqctl export_definitions - | jq .   # stream to stdout

rabbitmqctl import_definitions /tmp/rabbitmq-backup.json

curl -s http://localhost:15672/api/definitions -u guest:guest | rabbitmqctl import_definitions -

rabbitmqctl cluster_status

rabbitmqctl cluster_status --formatter=json | jq .partitions

rabbitmqctl stop_app && rabbitmqctl join_cluster rabbit@rabbit2 && rabbitmqctl start_app

rabbitmqctl join_cluster --ram rabbit@rabbit1   # RAM node (not recommended)

rabbitmqctl forget_cluster_node rabbit@dead-node

rabbitmqctl forget_cluster_node --offline rabbit@dead-node

rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl start_app

rabbitmq-plugins enable rabbitmq_management

rabbitmq-plugins enable rabbitmq_shovel rabbitmq_shovel_management

rabbitmq-plugins enable rabbitmq_stream rabbitmq_stream_management

rabbitmq-plugins list

rabbitmq-plugins list --enabled

rabbitmq-diagnostics check_running && echo "healthy"

rabbitmq-diagnostics check_port_connectivity && rabbitmq-diagnostics check_running

rabbitmq-diagnostics memory_breakdown

rabbitmq-diagnostics memory_breakdown --formatter=json | jq .

rabbitmq-diagnostics check_port_connectivity

rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}' --apply-to queues

rabbitmqctl set_policy ttl-1h "^transient\." '{"message-ttl":3600000}' --apply-to queues

rabbitmqctl set_policy max-length "^bounded\." '{"max-length":10000,"overflow":"reject-publish-dlx"}' --apply-to queues

rabbitmqctl eval 'rabbit_nodes:all_running().'

rabbitmqctl eval 'length(rabbit_amqqueue:list()).'

rabbitmqctl eval 'mnesia:system_info(all).'

# Declare: channel.exchangeDeclare("events", "direct", true)

# Bind:    channel.queueBind("order-created-queue", "events", "order.created")

# Publish: channel.basicPublish("events", "order.created", null, body)

# Binding "*.error" catches order.error, payment.error but not order.payment.error

# Binding "order.#" catches order.created, order.item.added, order.fulfilled

# Binding "#" catches everything (same as fanout but slower)

# Declare: channel.exchangeDeclare("broadcast", "fanout", true)

# Each subscriber declares its own queue and binds it — no routing key needed

# Publish: channel.basicPublish("broadcast", "", null, body)  // key ignored

# Bind with: {"format":"pdf","type":"report","x-match":"all"}

# Matches messages with BOTH format=pdf AND type=report headers

# channel.basicPublish("", "my-queue", null, body) sends direct to my-queue

# Declare source queue with DLX:

# channel.queueDeclare("orders", true, false, false, {"x-dead-letter-exchange": "dlx", "x-dead-letter-routing-key": "orders.dead"})

# Bind a "parking lot" queue to dlx with routing key "orders.dead"

# Declare exchange with AE:

# channel.exchangeDeclare("main", "topic", true, false, {"alternate-exchange": "unrouted"})

# channel.exchangeDeclare("unrouted", "fanout", true)

# channel.queueDeclare("task-queue", true, false, false, null)

# durable=true means the queue survives broker restart

# channel.queueDeclare("orders", true, false, false, {"x-queue-type": "quorum"})

# Set leader election strategy via policy: {"queue-leader-locator": "balanced"}

# channel.queueDeclare("events", true, false, false, {"x-queue-type": "stream", "x-max-age": "7D", "x-stream-max-segment-size-bytes": 536870912})

# Consumer subscribes with offset: {"x-stream-offset": "first"} or {"x-stream-offset": <timestamp>}

# {"x-message-ttl": 60000}  // 60 seconds

# Or via policy: rabbitmqctl set_policy ttl "^cache\." '{"message-ttl":5000}' --apply-to queues

# AMQP props: { expiration: "30000" }  // 30 seconds

# Note: expiration is a STRING in the AMQP spec, not an integer

# {"x-expires": 3600000}  // queue disappears after 1 hour idle

# channel.confirmSelect();

# channel.waitForConfirmsOrDie(5000);  // Java: wait up to 5s for all outstanding acks

# async: channel.addConfirmListener((tag, multiple) => { /* handle ack */ }, (tag, multiple) => { /* handle nack */ })

# channel.basicAck(deliveryTag, false)

# channel.basicNack(deliveryTag, false, false)  // nack, no requeue → DLX

# channel.basicReject(deliveryTag, false)       // reject single message

# channel.basicQos(10)  // at most 10 unacked messages per channel

# channel.basicQos(1, true)  // prefetch=1, global=true (applies per-channel not per-consumer)

# queue: durable=true

# message props: deliveryMode=2

# AMQP props: { deliveryMode: 2, contentType: "application/json" }

# {"x-max-length": 10000, "x-overflow": "reject-publish"}

# Via policy: rabbitmqctl set_policy bounded "^bnd\." '{"max-length":5000,"overflow":"reject-publish-dlx"}' --apply-to queues

# {"x-max-priority": 5}

# Publish high-priority: { priority: 5 }

# Publish normal: { priority: 1 }

# exclusive=true: channel.queueDeclare("", false, true, true, null)  // "" = server-generated name

# auto-delete=true: channel.queueDeclare("reply-12345", false, false, true, null)

# Declare one shared durable queue

# Start N worker processes each calling channel.basicConsume("tasks", ...)

# Set prefetch=1 to prevent work imbalance: channel.basicQos(1)

# Publisher: channel.basicPublish("events-fanout", "", null, body)

# Subscriber A: queueDeclare("", ..., true, true, ...) → queueBind(q, "events-fanout", "")

# Subscriber B: queueDeclare("", ..., true, true, ...) → queueBind(q, "events-fanout", "")

# channel.exchangeDeclare("events", "topic", true)

# Bind error queue: channel.queueBind("errors", "events", "#.error")

# Bind order queue: channel.queueBind("order-audit", "events", "order.#")

# Publish: channel.basicPublish("events", "order.payment.failed", null, body)

# Client: channel.basicPublish("", "rpc-server-queue", {replyTo: "amq.rabbitmq.reply-to", correlationId: uuid}, body)

# Use "amq.rabbitmq.reply-to" as the reply-to address for the fast direct reply protocol

# Server: on message, channel.basicPublish("", msg.properties.replyTo, {correlationId: msg.properties.correlationId}, responseBody)

# 1. Declare DLX: channel.exchangeDeclare("dlx", "direct", true)

# 2. Bind parking lot: channel.queueDeclare("parking-lot", true, ...); channel.queueBind("parking-lot", "dlx", "parking")

# 3. Source queue: {"x-dead-letter-exchange": "dlx", "x-dead-letter-routing-key": "parking"}

# 4. When consumer nacks with requeue=false, message lands in parking-lot

# wait-1s-queue: {"x-dead-letter-exchange": "main-exchange", "x-message-ttl": 1000}

# wait-10s-queue: {"x-dead-letter-exchange": "main-exchange", "x-message-ttl": 10000}

# After max retries, nack to a final parking-lot queue instead

# Install: rabbitmq-plugins enable rabbitmq_delayed_message_exchange

# channel.exchangeDeclare("delayed", "x-delayed-message", true, false, {"x-delayed-type": "direct"})

# Publish with header: {"x-delay": 5000}  // delay 5 seconds

# Enable: rabbitmq-plugins enable rabbitmq_shovel rabbitmq_shovel_management

# Set shovel via policy or management API: PUT /api/parameters/shovel/%2F/my-shovel

# {"src-uri":"amqp://","src-queue":"old-queue","dest-uri":"amqp://remote-host","dest-queue":"new-queue"}

# 1. Receive message with unique messageId in properties

# 2. if redis.set(messageId, "1", NX, EX, 86400): process and ack

# 3. else: ack without processing (already handled)

# 1. BEGIN TRANSACTION

# 2. UPDATE orders SET status="paid" WHERE id=123

# 3. INSERT INTO outbox(event_type, payload) VALUES ("order.paid", "...")

# 4. COMMIT

# 5. Poller: SELECT * FROM outbox WHERE published_at IS NULL; → publish; UPDATE outbox SET published_at=NOW()

# {"x-max-length": 1000, "x-overflow": "reject-publish"}

# Publisher must handle nack: pause publishing or fall back to local buffer

# {"x-queue-type": "stream", "x-max-age": "30D"}

# Subscribe from beginning: consumer args {"x-stream-offset": "first"}

# Subscribe from timestamp: consumer args {"x-stream-offset": <unix-timestamp-ms>}

# rabbitmqctl add_vhost /green

# rabbitmqctl set_permissions -p /green deploy ".*" ".*" ".*"

# Configure app to use AMQP_VHOST=/green for the new deployment

# Wrong: channel.basicConsume("tasks", true, callback)  // autoAck=true

# Right: channel.basicConsume("tasks", false, callback) // manual ack

# In callback: try { process(msg); channel.basicAck(tag, false); } catch { channel.basicNack(tag, false, false); }

# channel.basicQos(10)  // then channel.basicConsume(...)

# Wrong: ignore the error and keep using the same channel object

# Right: listen for channel.close event; create a new channel in the handler

# channel.on("error", (err) => { channel = connection.createChannel(); })

# Remove ha-all policy: rabbitmqctl clear_policy ha-all

# Migrate: drain classic queue, delete it, redeclare with x-queue-type: quorum

# Wrong: channel.basicPublish("ex", "key", null, body)  // default deliveryMode=1

# Right: channel.basicPublish("ex", "key", {deliveryMode: 2}, body)

# Temporary relief: rabbitmqctl set_vm_memory_high_watermark 0.6

# Check current: rabbitmq-diagnostics memory_breakdown

# Long-term: add RAM, enable lazy queues (page to disk), or reduce queue depth

# Wrong: create a new connection per message (serverless anti-pattern)

# Right: create one connection at startup; reuse across all handlers

# Check: rabbitmqctl list_connections | wc -l

# Node.js amqplib: amqp.connect("amqp://localhost?heartbeat=60")

# Java Spring AMQP: factory.setRequestedHeartbeat(60)

# Check ghost conns: rabbitmqctl list_connections name state  // look for "blocking" with no traffic

# channel.basicPublish("events", "wrong.key", {mandatory: true}, body)

# channel.on("return", (msg) => { console.error("Unroutable:", msg.fields.routingKey) })

# Or configure AE: {"alternate-exchange": "unrouted-ae"}

# Wrong: channel.basicNack(tag, false, true)  // will loop on poison message

# Right: check x-death header count; after 3 retries: channel.basicNack(tag, false, false)

# Minimum DLX setup:

# channel.exchangeDeclare("dlx", "fanout", true)

# channel.queueDeclare("dead-letters", true, false, false, null)

# channel.queueBind("dead-letters", "dlx", "")

# source queue: {"x-dead-letter-exchange": "dlx"}

# Wrong: share one channel across multiple threads

# Right: one connection → N channels (one per thread/goroutine/coroutine)

# Avoid: channel.txSelect(); channel.basicPublish(...); channel.txCommit();

# Use: channel.confirmSelect(); channel.basicPublish(...); channel.waitForConfirmsOrDie();

// Bad: only handles channel errors

channel.on('error', (err) => console.error('channel error', err));

// Good: handle both

connection.on('error', (err) => { reconnect(); });

channel.on('error', (err) => { reopenChannel(); });