再一次把RabbitMQ的架构图拿到这里来：
image.png
其主体分为三个大分部：
- RabbitMQ Server：它是一种传输服务，用于维护数据从生产者到消费者间的路线，保证数据能按指定方式进行传输。其内包括了交换机和队列，交换机分发生产者的数据到指定的队列，队列存放生产者发送的数据。
- Client A，B：生产者，产生数据
- Client 1，2，3：消费者，数据的接收方。
一个简单的RabbitMQ编程实例
生产者代码实现步骤：
- 获得与RabbitMQ Server的连接对象
- 通过连接对象获得Channel对象，Channel提供了与RabbitMQ交互的操作
- 连接用户储存消息数据的队列，不存在则创建队列
- 发送消息到指定的RabbitMQ队列中
- 关闭RabbitMQ代理的连接
只需按照上面步骤进行编写代码即可
import pika
class RabbitMQ(object):
def __init__(self, host, port, username, password, vhost):
self._host = host # broker IP
self._port = port # broker port
self._vhost = vhost # vhost
self._credentials = pika.PlainCredentials(username, password)
self._connection = None
def connect(self):
# 连接RabbitMQ的参数对象
parameter = pika.ConnectionParameters(self._host, self._port, self._vhost,
self._credentials, heartbeat_interval=10)
self._connection = pika.BlockingConnection(parameter) # 建立连接
def put(self, message_str, queue_name, route_key, exchange=''):
if self._connection is None:
return
channel = self._connection.channel() # 获取channel
channel.queue_declare(queue=queue_name) # 申明使用的queue
# 调用basic_publish方法向RabbitMQ发送数据， 这个方法应该只支持str类型的数据
channel.basic_publish(
exchange=exchange, # 指定exchange
routing_key=route_key, # 指定路由
body=message_str # 具体发送的数据
)
def getting_start(self, queue_name):
if self._connection is None:
return
channel = self._connection.channel()
channel.queue_declare(queue=queue_name)
# 调用basic_consume方法，可以传入一个回调函数
channel.basic_consume(self.callback,
queue=queue_name,
no_ack=True)
channel.start_consuming() # 相当于run_forever(), 当Queue中没有数据，则一直阻塞等待
@staticmethod
def callback(ch, method, properties, message_str):
"""定义一个回调函数"""
print "[x] Received {0}".format(message_str)
def close(self):
"""关闭RabbitMQ的连接"""
if self._connection is not None:
self._connection.close()
