RabbitMQ

Component format

To set up RabbitMQ pub/sub, create a component of type pubsub.rabbitmq. See the pub/sub broker component file to learn how ConsumerID is automatically generated. Read the How-to: Publish and Subscribe guide on how to create and apply a pub/sub configuration.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: rabbitmq-pubsub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
  - name: connectionString
    value: "amqp://localhost:5672"
  - name: protocol
    value: amqp  
  - name: hostname
    value: localhost 
  - name: username
    value: username
  - name: password
    value: password  
  - name: consumerID
    value: channel1
  - name: durable
    value: false
  - name: deletedWhenUnused
    value: false
  - name: autoAck
    value: false
  - name: deliveryMode
    value: 0
  - name: requeueInFailure
    value: false
  - name: prefetchCount
    value: 0
  - name: reconnectWait
    value: 0
  - name: concurrencyMode
    value: parallel
  - name: publisherConfirm
    value: false
  - name: enableDeadLetter # Optional enable dead Letter or not
    value: true
  - name: maxLen # Optional max message count in a queue
    value: 3000
  - name: maxLenBytes # Optional maximum length in bytes of a queue.
    value: 10485760
  - name: exchangeKind
    value: fanout
  - name: saslExternal
    value: false
  - name: ttlInSeconds
    value: 60
  - name: clientName
    value: {podName}
  - name: heartBeat
    value: 10s
  - name: publishMessagePropertiesToMetadata
    value: "true"

Spec metadata fields

Communication using TLS

To configure communication using TLS, ensure that the RabbitMQ nodes have TLS enabled and provide the caCert, clientCert, clientKey metadata in the component configuration. For example:

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: rabbitmq-pubsub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
  - name: host
    value: "amqps://localhost:5671"
  - name: consumerID
    value: myapp
  - name: durable
    value: false
  - name: deletedWhenUnused
    value: false
  - name: autoAck
    value: false
  - name: deliveryMode
    value: 0
  - name: requeueInFailure
    value: false
  - name: prefetchCount
    value: 0
  - name: reconnectWait
    value: 0
  - name: concurrencyMode
    value: parallel
  - name: publisherConfirm
    value: false
  - name: enableDeadLetter # Optional enable dead Letter or not
    value: true
  - name: maxLen # Optional max message count in a queue
    value: 3000
  - name: maxLenBytes # Optional maximum length in bytes of a queue.
    value: 10485760
  - name: exchangeKind
    value: fanout
  - name: saslExternal
    value: false
  - name: caCert
    value: ${{ myLoadedCACert }}
  - name: clientCert
    value: ${{ myLoadedClientCert }}
  - name: clientKey
    secretKeyRef:
      name: myRabbitMQClientKey
      key: myRabbitMQClientKey

Note that while the caCert and clientCert values may not be secrets, they can be referenced from a Dapr secret store as well for convenience.

Enabling message delivery retries

The RabbitMQ pub/sub component has no built-in support for retry strategies. This means that the sidecar sends a message to the service only once. When the service returns a result, the message will be marked as consumed regardless of whether it was processed correctly or not. Note that this is common among all Dapr PubSub components and not just RabbitMQ. Dapr can try redelivering a message a second time, when autoAck is set to false and requeueInFailure is set to true.

To make Dapr use more sophisticated retry policies, you can apply a retry resiliency policy to the RabbitMQ pub/sub component.

There is a crucial difference between the two ways to retry messages:

1. When using autoAck = false and requeueInFailure = true, RabbitMQ is the one responsible for re-delivering messages and any subscriber can get the redelivered message. If you have more than one instance of your consumer, then it’s possible that another consumer will get it. This is usually the better approach because if there’s a transient failure, it’s more likely that a different worker will be in a better position to successfully process the message.
2. Using Resiliency makes the same Dapr sidecar retry redelivering the messages. So it will be the same Dapr sidecar and the same app receiving the same message.

Create a RabbitMQ server

• Self-Hosted
• Kubernetes

docker run -d --hostname my-rabbit --name some-rabbit rabbitmq:3

helm install rabbitmq stable/rabbitmq

Use topic exchange to route messages

Setting exchangeKind to "topic" uses the topic exchanges, which are commonly used for the multicast routing of messages. In order to route messages using topic exchange, you must set the following metadata:

• routingKey:
  Messages with a routing key are routed to one or many queues based on the routing key defined in the metadata when subscribing.

• queueName:
  If you don’t set the queueName, only one queue is created, and all routing keys will route to that queue. This means all subscribers will bind to that queue, which won’t give the desired results.

For example, if an app is configured with a routing key keyA and queueName of queue-A:

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: orderspubsub
spec:
  topic: B
  routes: 
    default: /B
  pubsubname: pubsub
  metadata:
    routingKey: keyA
    queueName: queue-A

It will receive messages with routing key keyA, and messages with other routing keys are not received.

// publish messages with routing key `keyA`, and these will be received by the above example.
client.PublishEvent(context.Background(), "pubsub", "B", []byte("this is a message"), dapr.PublishEventWithMetadata(map[string]string{"routingKey": "keyA"}))
// publish messages with routing key `keyB`, and these will not be received by the above example.
client.PublishEvent(context.Background(), "pubsub", "B", []byte("this is another message"), dapr.PublishEventWithMetadata(map[string]string{"routingKey": "keyB"}))

Bind multiple routingKey

Multiple routing keys can be separated by commas.
The example below binds three routingKey: keyA, keyB, and "". Note the binding method of empty keys.

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: orderspubsub
spec:
  topic: B
  routes: 
    default: /B
  pubsubname: pubsub
  metadata:
    routingKey: keyA,keyB,

For more information see rabbitmq exchanges.

Use a consistent hash exchange for partitioned ordering

Setting exchangeKind to "x-consistent-hash" partitions a topic across consumers while preserving the order of messages that share a routing key. Every message for a given key is routed to the same queue, so that key is processed in order, while other keys are processed in parallel on other queues.

The routingKey metadata means something different on each side of publishing/subscribing:

• When publishing, it is the partition key. The exchange hashes it to choose a queue, so any messages that must stay in order relative to one another must share a key.
• When subscribing, it is the queue’s bucket weight. A single positive integer that represents the number of buckets the queue occupies on the hash ring, so the key space is divided between the bound queues in proportion to their weights. Unlike a topic exchange, a consistent hash exchange has one effective binding per queue, so only one value is accepted here.

Each consumer binds its own queue, so give each one a distinct queueName on the Subscription (or a distinct consumerID on the Component), and set concurrencyMode to single so that each partition is processed in order:

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: order-pub-sub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
  - name: connectionString
    value: "amqp://localhost:5672"
  - name: exchangeKind
    value: "x-consistent-hash"
  - name: concurrencyMode
    value: "single"

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: orderspubsub
spec:
  topic: orders
  routes:
    default: /orders
  pubsubname: order-pub-sub
  metadata:
    queueName: orders-partition-0
    routingKey: "10"

Publish with the partition key in the routingKey metadata. In each example both messages carry the same partition key, so both are routed to the same queue and processed in the order they were published.

• Python
• JavaScript
• Go

with DaprClient() as client:
    client.publish_event(
        pubsub_name='order-pub-sub',
        topic_name='orders',
        data='created',
        publish_metadata={'routingKey': 'tenant-42/serial-7'},
    )
    client.publish_event(
        pubsub_name='order-pub-sub',
        topic_name='orders',
        data='shipped',
        publish_metadata={'routingKey': 'tenant-42/serial-7'},
    )

const client = new DaprClient();

await client.pubsub.publish("order-pub-sub", "orders", "created", {
  metadata: { routingKey: "tenant-42/serial-7" },
});
await client.pubsub.publish("order-pub-sub", "orders", "shipped", {
  metadata: { routingKey: "tenant-42/serial-7" },
});

client.PublishEvent(context.Background(), "order-pub-sub", "orders", []byte("created"), dapr.PublishEventWithMetadata(map[string]string{"routingKey": "tenant-42/serial-7"}))
client.PublishEvent(context.Background(), "order-pub-sub", "orders", []byte("shipped"), dapr.PublishEventWithMetadata(map[string]string{"routingKey": "tenant-42/serial-7"}))

Use an externally managed topology

By default the component declares the topology it needs: an exchange named after the topic, a queue for each subscription, and the binding between them. If your exchanges and queues are instead managed declaratively, for instance by the RabbitMQ Cluster Kubernetes Operator’s Topology Operator, by Terraform, or by any other provisioning step, that declaration will collide.

An AMQP declaration is idempotent only when every property matches. Where the two disagree, RabbitMQ returns PRECONDITION_FAILED (406) and the component cannot use that topic at all. Because the defaults on each side differ, this happens out of the box rather than only in unusual configurations.

To avoid this collision, set either mode to "passive" to make the external owner the single source of truth:

• exchangeDeclareMode: passive asserts that the topic exchange exists and never creates or modifies it.
• queueDeclareMode: passive asserts that the consumer queue exists, and leaves queue.bind to the external owner, since a binding belongs to whoever owns the queue.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: order-pub-sub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
  - name: connectionString
    value: "amqp://localhost:5672"
  - name: exchangeDeclareMode
    value: "passive"
  - name: queueDeclareMode
    value: "passive"

Dead lettering follows queueDeclareMode, not exchangeDeclareMode, because the dead letter exchange and queue are named after the consumer at runtime: they belong to the queue rather than to the topology an external owner defines up front. With queueDeclareMode: declare the component creates them as before. With queueDeclareMode: passive it creates nothing and enableDeadLetter has no effect, so dead lettering becomes part of the queue definition, set up by whoever owns the queue and under whatever names they choose.

exchangeDeclareMode and queueDeclareMode are independent, so set whichever matches what the external owner actually manages. A common combination is an externally managed exchange with queues the component still declares, which is what a consistent hash setup needs when consumer queue names are only known at runtime:

  - name: exchangeDeclareMode
    value: "passive"
  - name: exchangeKind
    value: "x-consistent-hash"

Use priority queues

Dapr supports RabbitMQ priority queues. To set a priority for a queue, use the maxPriority topic subscription metadata.

Declarative priority queue example

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: pubsub
spec:
  topic: checkout
  routes: 
    default: /orders
  pubsubname: order-pub-sub
  metadata:
    maxPriority: 3

Programmatic priority queue example

• Python
• JavaScript
• Go

@app.route('/dapr/subscribe', methods=['GET'])
def subscribe():
    subscriptions = [
      {
        'pubsubname': 'pubsub',
        'topic': 'checkout',
        'routes': {
          'default': '/orders'
        },
        'metadata': {'maxPriority': '3'}
      }
    ]
    return jsonify(subscriptions)

const express = require('express')
const bodyParser = require('body-parser')
const app = express()
app.use(bodyParser.json({ type: 'application/*+json' }));

const port = 3000

app.get('/dapr/subscribe', (req, res) => {
  res.json([
    {
      pubsubname: "pubsub",
      topic: "checkout",
      routes: {
        default: '/orders'
      },
      metadata: {
        maxPriority: '3'
      }
    }
  ]);
})

package main

	"encoding/json"
	"net/http"

const appPort = 3000

type subscription struct {
	PubsubName string            `json:"pubsubname"`
	Topic      string            `json:"topic"`
	Metadata   map[string]string `json:"metadata,omitempty"`
	Routes     routes            `json:"routes"`
}

type routes struct {
	Rules   []rule `json:"rules,omitempty"`
	Default string `json:"default,omitempty"`
}

// This handles /dapr/subscribe
func configureSubscribeHandler(w http.ResponseWriter, _ *http.Request) {
	t := []subscription{
		{
			PubsubName: "pubsub",
			Topic:      "checkout",
			Routes: routes{
				Default: "/orders",
			},
      Metadata: map[string]string{
        "maxPriority": "3"
      },
		},
	}

	w.WriteHeader(http.StatusOK)
	json.NewEncoder(w).Encode(t)
}

Setting a priority when publishing a message

To set a priority on a message, add the publish metadata key maxPriority to the publish endpoint or SDK method.

• HTTP API (Bash)
• Python
• JavaScript
• Go

curl -X POST http://localhost:3601/v1.0/publish/order-pub-sub/orders?metadata.priority=3 -H "Content-Type: application/json" -d '{"orderId": "100"}'

with DaprClient() as client:
        result = client.publish_event(
            pubsub_name=PUBSUB_NAME,
            topic_name=TOPIC_NAME,
            data=json.dumps(orderId),
            data_content_type='application/json',
            metadata= { 'priority': '3' })

await client.pubsub.publish(PUBSUB_NAME, TOPIC_NAME, orderId, { 'priority': '3' });

client.PublishEvent(ctx, PUBSUB_NAME, TOPIC_NAME, []byte(strconv.Itoa(orderId)), map[string]string{"priority": "3"})

Use quorum queues

By default, Dapr creates classic queues. To create quorum queues, add the following metadata to your pub/sub subscription

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: pubsub
spec:
  topic: checkout
  routes: 
    default: /orders
  pubsubname: order-pub-sub
  metadata:
    queueType: quorum

Time-to-live

You can set a time-to-live (TTL) value at either the message or component level. Set default component-level TTL using the component spec ttlInSeconds field in your component.

Single Active Consumer

The RabbitMQ Single Active Consumer setup ensures that only one consumer at a time processes messages from a queue and switches to another registered consumer if the active one is canceled or fails. This approach might be required when it is crucial for messages to be consumed in the exact order they arrive in the queue and if distributed processing with multiple instances is not supported. When this option is enabled on a queue by Dapr, an instance of the Dapr runtime will be the single active consumer. To allow another application instance to take over in case of failure, Dapr runtime must probe the application’s health and unsubscribe from the pub/sub component.

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: pubsub
spec:
  topic: orders
  routes:
    default: /orders
  pubsubname: order-pub-sub
  metadata:
    singleActiveConsumer: "true"

Publishing message properties to metadata

To enable message properties being published in the metadata, set the publishMessagePropertiesToMetadata field to "true" in the component spec. This will include properties such as message ID, timestamp, and headers in the metadata of the published message.

Related links

• Basic schema for a Dapr component in the Related links section
• Read this guide for instructions on configuring pub/sub components
• Pub/Sub building block