Strictly speaking, AMQP (and 0.9.1 version used) is a protocol, not a full-fledged driver, so you can use any servers that support this protocol (on your own, only rabbitmq was tested), such as:
However, it is recommended to use RabbitMQ as the main implementation, and reliable performance with other implementations is not guaranteed.
To install and configure the RabbitMQ, use the corresponding documentation page.
Every message pushed to the RabbitMQ server uses publisher confirms. A message is considered sent only after the server confirms it. This is a reliable way to ensure delivery to the server.
After that, you should configure the connection to the server in the amqp section. This configuration section contains exactly one addr key with a connection DSN. The TLS configuration sits in the amqp.tls section and consists of the following options:
key: path to a key file.
cert: path to a certificate file.
root_ca: path to Root CAs used by the AMQP client to trust and verify the broker/server certificate during TLS dial.
client_auth_type: also known as mTLS. Possible values are: no_client_cert, request_client_cert, require_any_client_cert, verify_client_cert_if_given, require_and_verify_client_cert.
You should also configure rabbitMQ with TLS support: link.
.rr.yaml
amqp:addr:amqp://guest:[email protected]:5672 # AMQPS TLS configuration # # This section is optionaltls: # Path to the key file # # This option is requiredkey:"" # Path to the certificate # # This option is requiredcert:"" # Path to Root CAs used by the AMQP client to trust and verify the broker/server certificate during TLS dial. # # This option is optionalroot_ca:"" # Client auth type (mTLS, peer verification). # # This option is optional. Default value: no_client_cert. Possible values: no_client_cert, request_client_cert, require_any_client_cert, verify_client_cert_if_given, require_and_verify_client_certclient_auth_type:no_client_cert
Upon establishing a connection to the server, you can create a new queue that utilizes this connection and encompasses the queue settings, including those specific to AMQP.
Configuration
Configuration options
Here is a detailed description of each of the amqp-specific options:
Priority
priority- job priority. A lower value corresponds to a higher priority. For instance, consider two pipelines: pipe1 with a priority of 1 and pipe10 with a priority of 10. Workers will only take jobs from pipe10 if all the jobs from pipe1 have been processed.
Prefetch
prefetch - rabbitMQ QoS prefetch. See also "prefetch-size". Note that if you use a large number of workers and a small prefetch number, some of the workers may not be loaded with messages (jobs) due to the blocking nature of the prefetch. This would result in poor RoadRunner performance and waste of resources.
Queue
queue - AMQP internal (inside the driver) queue name. Optional for producer-only pipelines, required for consumer lifecycle operations (run, resume, pause).
Producer-only pipeline: queue can be empty, push works, but run, resume, and pause will fail without a queue.
Exchange type
exchange_type - rabbitMQ exchange type. May be one of direct, fanout, topic, headers.
Routing key
routing_key - queue's routing key. Required for push when exchange_type != fanout.
Exclusive
exclusive - applied to the queue, exclusive queues cannot be redeclared. If set to true, and you attempt to declare the same pipeline twice, it will result in an error.
Multiple ack
multiple_ack - this delivery, along with all prior unacknowledged deliveries on the same channel, will be acknowledged. This feature is beneficial for batch processing of deliveries and is applicable only for Ack, not for Nack.
Requeue on fail
requeue_on_fail - requeue on Nack (by RabbitMQ).
Read more about Nack in RabbitMQ official docs: https://www.rabbitmq.com/confirms.html#consumer-nacks-requeue
Queue headers
queue_headers - used to pass arguments to the Queue create method, such as x-queue-mode: lazy
Durable
durable - create a durable queue.
Default: false
Delete queue on stop
delete_queue_on_stop - delete the queue when the pipeline is stopped.
Default: false
Redial timeout
redial_timeout - Redial timeout (in seconds). How long to try to reconnect to the AMQP server.
version: "3"
amqp:
addr: amqp://guest:[email protected]:5672
# AMQPS TLS configuration
#
# This section is optional
tls:
# Path to the key file
#
# This option is required
key: ""
# Path to the certificate
#
# This option is required
cert: ""
# Path to Root CAs used by the AMQP client to trust and verify the broker/server certificate during TLS dial.
#
# This option is optional
root_ca: ""
# Client auth type (mTLS, peer verification).
#
# This option is optional. Default value: no_client_cert. Possible values: no_client_cert, request_client_cert, require_any_client_cert, verify_client_cert_if_given, require_and_verify_client_cert
client_auth_type: no_client_cert
jobs:
pipelines:
# User defined name of the queue.
example:
# Driver name
#
# This option is required.
driver: amqp
# Driver's configuration
#
# Should not be empty
config:
# QoS - prefetch.
#
# Default: 10
prefetch: 10
# Pipeline priority
#
# If the job has priority set to 0, it will inherit the pipeline's priority. Default: 10.
priority: 1
# Redial timeout (in seconds). How long to try to reconnect to the AMQP server.
#
# Default: 60
redial_timeout: 60
# Durable queue
#
# Default: false
durable: false
# Durable exchange (rabbitmq option: https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchanges)
#
# Default: false
exchange_durable: false
# Auto-delete (exchange is deleted when last queue is unbound from it): https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchanges
#
# Default: false
exchange_auto_delete: false
# Auto-delete (queue that has had at least one consumer is deleted when last consumer unsubscribes) (rabbitmq option: https://www.rabbitmq.com/queues.html#properties)
#
# Default: false
queue_auto_delete: false
# Delete queue when stopping the pipeline
#
# Default: false
delete_queue_on_stop: false
# Queue name
#
# Optional for producer-only pipelines. Required for run/resume/pause.
# Can be omitted for push-only mode.
queue: test-1-queue
# Exchange name
#
# Optional. Default: amqp.default
exchange: amqp.default
# Exchange type
#
# Default: direct. Possible values: direct, fanout, topic, headers.
exchange_type: direct
# Routing key for the queue
#
# Default: empty. Required for push when exchange_type != fanout.
routing_key: test
# Declare a queue exclusive at the exchange
#
# Default: false
exclusive: false
# When multiple is true, this delivery and all prior unacknowledged deliveries
# on the same channel will be acknowledged. This is useful for batch processing
# of deliveries
#
# Default: false
multiple_ack: false
# The consumer_id is identified by a string that is unique and scoped for all consumers on this channel.
#
# Default: "roadrunner" + uuid.
consumer_id: "roadrunner-uuid"
# Use rabbitmq mechanism to requeue the job on fail
#
# Default: false
requeue_on_fail: false
# Queue headers (new in 2.12.2)
#
# Default: null
queue_headers:
x-queue-mode: lazy
