介绍
RabbitMQ 是一个开源的消息代理和队列服务器，实现高级消息队列协议 (AMQP)。它可以在多语言环境中作为中间件来处理应用程序之间的消息传递。
安装与配置
安装 RabbitMQ
1. 访问 [RabbitMQ 下载页面](https://www.rabbitmq.com/download.html) 并下载适合您操作系统的版本。
2. 安装并启动服务。
配置环境
- 对于开发环境，通常不需要特别配置。
- 可以通过命令行工具 `rabbitmqctl` 进行管理。
快速开始
创建虚拟主机
```bash
rabbitmqctl add_vhost my_vhost
```
用户创建及权限设置
```bash
rabbitmqctl add_user my_user my_password
rabbitmqctl set_permissions -p my_vhost my_user ".*" ".*" ".*"
```
启用插件
```bash
rabbitmq-plugins enable rabbitmq_management
```
生产者示例
生产者负责发送消息到队列中。
Python 示例
```python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='',
routing_key='hello',
body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()
```
消费者示例
消费者从队列中读取消息。
Python 示例
```python
import pika
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.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()
```
发布/订阅模式
发布/订阅模式允许多个消费者接收来自同一个生产者的消息。
创建交换机
```python
channel.exchange_declare(exchange='logs', exchange_type='fanout')
```
生产者
```python
channel.basic_publish(exchange='logs',
routing_key='',
body='Hello Logs!')
```
消费者
```python
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange='logs', queue=queue_name)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r" % body)
channel.basic_consume(queue=queue_name,
on_message_callback=callback,
auto_ack=True)
channel.start_consuming()
```
工作队列
工作队列可以将任务分发给多个工作者处理。
生产者
```python
channel.queue_declare(queue='task_queue', durable=True)
message = 'A message'
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))
print(" [x] Sent %r" % message)
```
消费者
```python
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue',
on_message_callback=callback)
channel.start_consuming()
```
路由模式
路由模式允许根据消息中的路由键来选择队列。
创建交换机
```python
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
```
生产者
```python
severities = ['info', 'warning', 'error']
for severity in severities:
message = f'Log message: {severity}'
channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)
print(f" [x] Sent {message}")
```
消费者
```python
severities = ['info', 'warning']
for severity in severities:
queue_name = f'{severity}_queue'
channel.queue_declare(queue=queue_name)
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
for severity in severities:
queue_name = f'{severity}_queue'
channel.basic_consume(queue=queue_name,
on_message_callback=callback,
auto_ack=True)
channel.start_consuming()
```
主题模式
主题模式是基于通配符的路由模式。
创建交换机
```python
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
```
生产者
```python
routing_key = 'kern.critical'
message = 'Critical kernel alert'
channel.basic_publish(exchange='topic_logs',
routing_key=routing_key,
body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
```
消费者
```python
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
binding_keys = ['kern.*', '*.critical']
for binding_key in binding_keys:
channel.queue_bind(exchange='topic_logs',
queue=queue_name,
routing_key=binding_key)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(queue=queue_name,
on_message_callback=callback,
auto_ack=True)
channel.start_consuming()