消息系统允许软件应用相互连接和扩展．这些应用可以相互链接起来组成一个更大的应用，或者将用户设备和数据进行连接．消息系统通过将消息的发送和接收分离来实现应用程序的异步和解偶．
数据投递，非阻塞操作或推送通知、发布／订阅，异步处理，或者工作队列。所有这些都属于消息系统的模式。
消息系统的几个有点
- 异步（asynchronous）：耗时的工作可以直接丢给消费者，不会阻塞生产者
- 可扩展（scale）：消息机制的工作模式，在理论上可以让无限的消费者接入进来，使得横向扩展变得异常简单
- 模块化（modulize）：生产者和消费者不需要知道双方的存在，而且可以使用不同的语言和框架，在物理上位于不同的地方。
Paste_Image.png
Exchanges：生产者把消息发送到某个 exchange，exchange 的主要工具就是根据一定的规则把消息分发到不同的 queuesQueues：消息最终被发送到的地方，消费者也是从这里拿消息进行处理Routing key：queue 要绑定（binding）到某个 exchange 才有可能接受这个 exchange 转发过来的消息，它们之间的绑定要有 binding key（当 exchange 是 fanout 类型的时候并不需要）。然后每个消息发过来的时候，都会带着 routing key，根据 exchange 的类型，binding key 和 routing key，消息才能正确被转发。Connection：生产者和消费者需要连接到 rabbitmq 才能发送和读取消息，这个连接就是 connection，本质上就是 TCP 连接。Channel：为了减少网络负载，和减少 TCP 链接数。多个不同的 生产者可以在同一个 TCP 发送消息，只不过在这个 connection 下面独自的 channel 里。你可以把 channel 想象成一根网线里独立的细线。每个 channel 都有自己的 id，用来标识自己。虽然有 connection，不过所有的通信都是在对应的 channel 里进行的。vhost：Virtual host，是起到隔离作用的。每一个 vhost 都有自己的 exchanges 和 queues，它们互不影响。不同的应用可以跑在相同的 rabbitmq 上，使用 vhost 把它们隔离开就行。默认情况下，rabbitmq 安装后，默认的 vhost 是 /。
1.简单的生产者与消费者
send.py
\#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='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()
receive.py
\#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
print ' [*] Waiting for messages. To exit press CTRL+C'
def callback(ch, method, properties, body):
print " [x] Received %r" % (body,)
channel.basic_consume(callback,
queue='hello',
no_ack=True)
channel.start_consuming()
```
####2.工作队列
new_task.py
```python
\#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
message = ' '.join(sys.argv[1:]) or "Hello World!"
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()
```
worker.py
channel.basic_qos(prefetch_count=1) 保证每次取一个
channel.queue_declare(queue='task_queue', durable=True) 队列持久化
```python
\#!/usr/bin/env python
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
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,)
time.sleep( body.count('.') )
print " [x] Done"
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
queue='task_queue')
channel.start_consuming()
```
####3.发布／订阅
fanout 它把消息发送给它所知道的所有队列
emit_log.py
```python
#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
type='fanout')
message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs',
routing_key='',
body=message)
print " [x] Sent %r" % (message,)
connection.close()
```
receive_logs.py
当与消费者（consumer）断开连接的时候，这个队列应当被立即删除。exclusive 标识符即可达到此目的。
```
result = channel.queue_declare(exclusive=True)
```
```
\#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
type='fanout')
result = channel.queue_declare(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(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
```
===============================================================
3.路由(Routing)
在前面的教程中，我们实现了一个简单的日志系统。可以把日志消息广播给多个接收者。
本篇教程中我们打算新增一个功能 —— 使得它能够只订阅消息的一个字集。例如，我们只需要把严重的错误日志信息写入日志文件（存储到磁盘），但同时仍然把所有的日志信息输出到控制台中
#####绑定（Bindings）
队列绑定到路由
```
channel.queue_bind(exchange=exchange_name,
queue=queue_name)
```
绑定的时候可以带上一个额外的 routing_key 参数。
```
channel.queue_bind(exchange=exchange_name,
queue=queue_name,
routing_key='black')
```
直连交换机（Direct exchange）
emit_log_direct.py 的代码
```
\#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)
print " [x] Sent %r:%r" % (severity, message)
connection.close()
```
receive_logs_direct.py 的代码：
```
\#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = sys.argv[1:]
if not severities:
print >> sys.stderr, "Usage: %s [info] [warning] [error]" % \
(sys.argv[0],)
sys.exit(1)
for severity in severities:
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,)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
```
==============================================================
####4.主题交换机
当一个队列的绑定键为 "#"（井号） 的时候，这个队列将会无视消息的路由键，接收所有的消息。
当 * (星号) 和 # (井号) 这两个特殊字符都未在绑定键中出现的时候，此时主题交换机就拥有的直连交换机的行为。
emit_log_topic.py 的代码：
```
\#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
type='topic')
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
routing_key=routing_key,
body=message)
print " [x] Sent %r:%r" % (routing_key, message)
connection.close()
```
receive_logs_topic.py 的代码：
```
\#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
type='topic')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
binding_keys = sys.argv[1:]
if not binding_keys:
print >> sys.stderr, "Usage: %s [binding_key]..." % (sys.argv[0],)
sys.exit(1)
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(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
```