Python操作RabbitMQ

RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。

MQ全称为Message Queue，消息队列是一种应用程序对应程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接他们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

RabbitMQ的安装：

1、安装配置epel源

   $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm

2、安装erlang

   $ yum -y install erlang

3、安装RabbitMQ

   $ yum -y install rabbitmq-server

4、启动与停止：service rabbitmq-server start/stop

python安装API：

pip install pika  或者  easy_install pika  或者源码安装：

https://pypi.python.org/pypi/pika

对于RabbitMQ来说，生产和消费不再针对内存里的一个queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。

import pika

# 创建连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
channel = connection.channel() # 创建频道
channel.queue_declare(queue='hello') # 创建队列hello，若存在则忽略
# 向队列hello中发消息， body为发的消息内容
channel.basic_publish(exchange='', routing_key='hello', body='Hello World')
print('saf')
connection.close()

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
chanel = conn.channel()
chanel.queue_declare(queue='hello')


def callback(ch, method, properties, body):
    print(body)

# ch为channel， method为函数名字， property为属性， body为取到的消息
chanel.basic_consume(callback, queue='hello', no_ack=True)
# no_ack = False为，如果消费者遇到情况（its channel is closed，connection
# is closed，or TCP connection is lost）挂掉了，那么，RabbitMQ会重新将该任务添加到队列中
print('hehehehe')
chanel.start_consuming()

 1、acknowledgment 消息不丢失：no_ack = False，如果消费者遇到情况（its channel is closed,connection is closed,or TCP connection is lost）挂掉了，那么，RabbitMQ会重新将该任务添加到队列中。

import pika, time

conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
chanel = conn.channel()
chanel.queue_declare(queue='hello')


def callback(ch, method, properties, body):
    print('消息：', body)
    time.sleep(10)
    print('ok')
    ch.basic_ack(delivery_tag=method.delivery_tag)

chanel.basic_consume(callback, queue='hello', no_ack=False)
print('我是消费者，等待消息')
chanel.start_consuming()

2、durable（持久化） 消息不丢失：当RabbitMQ服务宕机，后也不用担心消息的丢失。

import pika
conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
channel = conn.channel()
channel.queue_declare(queue='hello1', durable=True)
# delivery_mode=2意思是做持久化
channel.basic_publish(exchange='', routing_key='hello1',
                      body='hello world',
                      properties=pika.BasicProperties(
                          delivery_mode=2,
                      ))
print('发送消息')
conn.close()

import pika, time

conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
chanel = conn.channel()
chanel.queue_declare(queue='hello', durable=True)


def callback(ch, method, properties, body):
    print('消息是：', body)
    time.sleep(10)
    ch.basic_ack(delivery_tag=method.delivery_tag)

chanel.basic_consume(callback, queue='hello1', no_ack=False)
print('等待消息中。。。')
chanel.start_consuming()

3、消息获取顺序：默认消息队列里的数据是按照吮吸被消费者拿走（默认消费者按照间隔）
有些时候不需要按照间隔去去任务更好一些。channel.basic(prefetch_count=1)表示谁来谁取，不再按照间隔

import pika
conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
channel = conn.channel()

channel.queue_declare(queue='hello2')


def callback(ch, method, properties, body):
    print('消息是', body)
    print('ok')
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback, queue='hello2', no_ack=False)
print('等待消息中')
channel.start_consuming()

4、发布和订阅

发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。exchange type=fanout

# 订阅者
import pika
conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
channel = conn.channel()
channel.exchange_declare(exchange='logs', type='fanout')
res = channel.queue_declare(exclusive=True)
queue_name = res.method.queue
print(queue_name)
channel.queue_bind(exchange='logs', queue=queue_name)
print('等待消息logs中')


def callback(ch, method, properties, body):
    print('消息是', body)

channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()

# 发布者
import pika, sys
conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
channel = conn.channel()
channel.exchange_declare(exchange='logs', type='fanout')
msg = ' '.join(sys.argv[1:]) or 'info: Hello World'
channel.basic_publish(exchange='logs', routing_key='',
                      body=msg)
print(msg)
conn.close()

5、关键字发送：发送消息时，明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据关键字判定应该将数据发送至指定队列。 exchange type=direct

import pika, sys
conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
channel = conn.channel()
channel.exchange_declare(exchange='direct_logs', type='direct')
res = channel.queue_declare(exclusive=True)
queue_name = res.method.queue
severities = sys.argv[1:]
if not severities:
    # sys.stderr.write(sys.argv[0])
    # sys.exit(1)
    severi = input('>>:').strip().split()
    severities = severi

for severity in severities:
    channel.queue_bind(exchange='direct_logs', queue=queue_name,
                       routing_key=severity)

print('等待消息')


def callback(ch, method, properties, body):
    print('收到消息：', method.routing_key, body)

channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()

import pika, sys

conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
channel = conn.channel()
channel.exchange_declare(exchange='direct_logs', type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
msg = ' '.join(sys.argv[2:]) or 'Hello the Sea.'
channel.basic_publish(exchange='direct_logs', routing_key=severity,
                      body=msg)
print('发送消息：', msg)
conn.close()

6、模糊匹配：在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange， exchange将传入‘路由值’和‘关键字’进行匹配，匹配成功，则将数据发送到指定队列。    exchange  type=topic
   #  表示可以匹配0个或多个单词

   *  表示只能匹配一个单词

import pika, sys
conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
channel = conn.channel()
channel.exchange_declare(exchange='topic_logs', type='topic')
res = channel.queue_declare(exclusive=True)
queue_name = res.method.queue
binding_keys = sys.argv[1:]
if not binding_keys:
    # sys.stderr.write(sys.argv[0])
    # sys.exit(1)
    binding_keys = input('>>:').strip().split()

for binding_key in binding_keys:
    channel.queue_bind(exchange='topic_logs', queue=queue_name,
                       routing_key=binding_key)
print('等待消息中。。。')


def callback(ch, method, properties, body):
    print(method.routing_key, body)

channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()

import pika, sys
conn = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.254.129'))
channel = conn.channel()
channel.exchange_declare(exchange='topic_logs', type='topic')
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'info.boy'
msg = ' '.join(sys.argv[2:]) or 'Hello the sky'
channel.basic_publish(exchange='topic_logs', routing_key=routing_key,
                      body=msg)
print(routing_key, msg)
conn.close()