RabbitMQ 入门教程
1. 引言
RabbitMQ 是一个开源的消息代理和队列服务器，实现了 AMQP 0-9-1 标准。本教程将指导你如何安装、配置和使用 RabbitMQ 进行消息传递。
2. 安装 RabbitMQ
2.1 安装 RabbitMQ 服务器
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 启动服务
```bash
sudo service rabbitmq-server start
```
2.3 配置管理插件
```bash
sudo rabbitmq-plugins enable rabbitmq_management
```
2.4 访问管理界面
- 浏览器访问: http://localhost:15672/
- 用户名/密码: guest/guest
3. 快速入门
3.1 创建 Python 虚拟环境
```bash
python3 -m venv venv
source venv/bin/activate
pip install pika
```
3.2 发送者 (sender.py)
```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()
```
3.3 接收者 (receiver.py)
```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()
```
4. 工作队列
工作队列允许我们将任务分配给多个工作者，实现负载均衡。
4.1 发送者 (work_queue_sender.py)
```python
import pika
import sys
message = ' '.join(sys.argv[1:]) or "Hello World!"
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
channel.basic_publish(
exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
))
print(" [x] Sent %r" % message)
connection.close()
```
4.2 接收者 (work_queue_receiver.py)
```python
import pika
import time
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(body.count(b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag=method.delivery_tag)
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.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. 发布/订阅模式
发布/订阅模式允许消息被广播到多个消费者。
5.1 发送者 (fanout_sender.py)
```python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.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)
connection.close()
```
5.2 接收者 (fanout_receiver.py)
```python
import pika
import uuid
def on_message(channel, method, props, body):
print(" [x] %r" % body)
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs', exchange_type='fanout')
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')
channel.basic_consume(queue=queue_name, on_message_callback=on_message, auto_ack=True)
channel.start_consuming()
```
6. 路由模式
路由模式可以根据特定的键来发送消息。
6.1 发送者 (direct_sender.py)
```python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.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))
connection.close()
```
6.2 接收者 (direct_receiver.py)
```python
import pika
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
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. 主题模式
主题模式允许更复杂的路由规则。
7.1 发送者 (topic_sender.py)
```python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
routing_key = 'kern.critical'
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))
connection.close()
```
7.2 接收者 (topic_receiver.py)
```python
import pika
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
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)
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()
标签：pika,exchange,入门教程,RabbitMQ,queue,connection,message,channel From： https://blog.csdn.net/qq_40698086/article/details/141553876