RabbitMQGetting started tutorial
1. Introduction
RabbitMQ is an open source message broker andqueueserver, implements the AMQP 0-9-1 standard. This tutorial will guide you on how to install, configure, and use RabbitMQ for messaging.
2. Install RabbitMQ
2.1 Installing the RabbitMQ server
2.1.1 Ubuntu/Debian
```bash
sudo apt-get update
sudo apt-get install rabbitmq-server
```
2.1.2 CentOS/RHEL
```bash
sudo yum install epel-release
sudo yum install rabbitmq-server
```
2.2 Start the service
```bash
sudo service rabbitmq-server start
```
2.3 Configuration Management Plug-in
```bash
sudo rabbitmq-plugins enable rabbitmq_management
```
2.4 Access management interface
- Browser access: http://localhost:15672/
- Username/password: guest/guest
3. Quick Start
3.1 Creating a Python virtual environment
```bash
python3 -m venv venv
source venv/bin/activate
pip install pika
```
3.2 Sender ()
```python
import pika
connection = (('localhost'))
channel = ()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='',
routing_key='hello',
body='Hello World!')
print(" [x] Sent 'Hello World!'")
()
```
3.3 Recipient ()
```python
import pika
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
connection = (('localhost'))
channel = ()
channel.queue_declare(queue='hello')
channel.basic_consume(queue='hello',
on_message_callback=callback,
auto_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
```
4. Work queue
Work queues allow us to assign tasks to multiple workers to achieve load balancing.
4.1 Sender (work_queue_sender.py)
```python
import pika
import sys
message = ' '.join([1:]) or "Hello World!"
connection = (('localhost'))
channel = ()
channel.queue_declare(queue='task_queue', durable=True)
channel.basic_publish(
exchange='',
routing_key='task_queue',
body=message,
properties=(
delivery_mode=2, # make message persistent
))
print(" [x] Sent %r" % message)
()
```
4.2 Recipient (work_queue_receiver.py)
```python
import pika
import time
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
((b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag=method.delivery_tag)
connection = (('localhost'))
channel = ()
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)
channel.start_consuming()
```
5. Publish/Subscribe Mode
The Publish/Subscribe mode allows messages to be broadcast to multiple consumers.
5.1 Sender (fanout_sender.py)
```python
import pika
connection = (('localhost'))
channel = ()
channel.exchange_declare(exchange='logs', exchange_type='fanout')
message = "Info: Hello World!"
channel.basic_publish(exchange='logs', routing_key='', body=message)
print(" [x] Sent %r" % message)
()
```
5.2 Recipient (fanout_receiver.py)
```python
import pika
import uuid
def on_message(channel, method, props, body):
print(" [x] %r" % body)
connection = (('localhost'))
channel = ()
channel.exchange_declare(exchange='logs', exchange_type='fanout')
result = channel.queue_declare(queue='', exclusive=True)
queue_name =
channel.queue_bind(exchange='logs', queue=queue_name)
print(' [*] Waiting for logs. To exit press CTRL+C')
channel.basic_consume(queue=queue_name, on_message_callback=on_message, auto_ack=True)
channel.start_consuming()
```
6. Routing mode
Routing mode can send messages based on specific keys.
6.1 Sender (direct_sender.py)
```python
import pika
connection = (('localhost'))
channel = ()
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
severity = 'info'
message = 'Info: Hello World!'
channel.basic_publish(exchange='direct_logs', routing_key=severity, body=message)
print(" [x] Sent %r:%r" % (severity, message))
()
```
6.2 Recipient (direct_receiver.py)
```python
import pika
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
connection = (('localhost'))
channel = ()
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
result = channel.queue_declare(queue='', exclusive=True)
queue_name =
severities = ['error', 'warning']
for severity in severities:
channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity)
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
print(' [*] Waiting for logs. To exit press CTRL+C')
channel.start_consuming()
```
7. Theme mode
Theme mode allows for more complex routing rules.
7.1 Sender (topic_sender.py)
```python
import pika
connection = (('localhost'))
channel = ()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
routing_key = ''
message = 'A critical kernel error'
channel.basic_publish(exchange='topic_logs', routing_key=routing_key, body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
()
```
7.2 Recipient (topic_receiver.py)
```python
import pika
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
connection = (('localhost'))
channel = ()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
result = channel.queue_declare(queue='', exclusive=True)
queue_name =
binding_keys = ['kern.*', '*.critical']
for binding_key in binding_keys:
channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key)
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
print(' [*] Waiting for logs. To exit press CTRL+C')
channel.start_consuming()