介绍
RabbitMQ 是一个开源的消息代理和队列服务器，实现高级消息队列协议 (AMQP)。它可以在多台机器之间共享任务，或者在服务出现问题时作为缓冲。
安装与配置
安装 RabbitMQ
1. Ubuntu:
```bash
sudo apt-get update
sudo apt-get install rabbitmq-server
```
2. MacOS (通过 Homebrew):
```bash
brew install rabbitmq
```
启动服务
```bash
rabbitmq-server
```
配置
编辑 `rabbitmq.config` 文件以进行配置:
```erlang
[
{rabbit, [
{loopback_users, []}
]}
].
```
基础概念
- Exchange - 消息交换器，决定消息发送到哪些队列。
- Queue - 消息队列，用于存储消息直到被消费。
- Binding - 绑定关系，将交换器与队列连接起来。
- Routing Key - 路由键，用于绑定交换器和队列。
Python 示例
创建环境
```bash
pip install pika
```
生产者
```python
import pika
def send_message(host: str, queue_name: str, message: str):
connection = pika.BlockingConnection(pika.ConnectionParameters(host=host))
channel = connection.channel()
channel.queue_declare(queue=queue_name)
channel.basic_publish(exchange='', routing_key=queue_name, body=message)
print(f" [x] Sent '{message}'")
connection.close()
if __name__ == '__main__':
send_message('localhost', 'hello', 'Hello World!')
```
消费者
```python
import pika
def callback(ch, method, properties, body):
print(f" [x] Received {body}")
def receive_messages(host: str, queue_name: str):
connection = pika.BlockingConnection(pika.ConnectionParameters(host=host))
channel = connection.channel()
channel.queue_declare(queue=queue_name)
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
if __name__ == '__main__':
receive_messages('localhost', 'hello')
```
发布/订阅模式
创建 Exchange
```python
channel.exchange_declare(exchange='logs', exchange_type='fanout')
```
生产者
```python
channel.basic_publish(exchange='logs', routing_key='', body=message)
```
消费者
```python
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange='logs', queue=queue_name)
channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
```
工作队列
生产者
```python
channel.queue_declare(queue='task_queue', durable=True)
channel.basic_publish(exchange='', routing_key='task_queue', body=message,
properties=pika.BasicProperties(delivery_mode=2,))
```
消费者
```python
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback, auto_ack=False)
