python操作RabbitMQ

首先官网也有相关demo可以参考：http://www.rabbitmq.com/getstarted.html

一、基本操作

需要准备一台RbbitMQ服务器。安装第三方模块pika。

producer端：

# -*- coding:utf-8 -*- 
# Author：Brownyangyang
import pika

#生成一个socket实例
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)

#声明一个管道
channel = connection.channel()

#声明一个叫number1的队列
channel.queue_declare(queue="number1")

#通过管道发送消息，number1是queue名，body是消息主体
channel.basic_publish(exchange='',
                      routing_key="number1",
                      body="hello world")

print("[x] sent 'hello World!'")
connection.close()

consumer端：

# -*- coding:utf-8 -*- 
# Author：Brownyangyang
import pika

#生成实例
connnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#声明一个管道
channel = connnection.channel()

#声明一个叫number1的队列,如果不声明,consumer如果在producer之前启动会报错
channel.queue_declare(queue='number1')

def callback(ch,method,properties,body):
    print("---》》》",ch,method,properties)
    print("[x] Received %r" % body)

##消费消息,如果收到消息，就调用callback函数
channel.basic_consume(callback,
                      queue='number1',
                      no_ack=True)  ##意味着客户端不管是否收到消息都不会通知服务端

print("[*] Waiting for messages.CTRL+C to exit.")

#开始监听消息队列
channel.start_consuming()

二、分发轮询

     默认情况下，开启 了多个consumer是轮询接收消息的，注意上面代码中channel.basic_consume方法有个no_ack=True，这个表示consumer端不管消息是否处理完毕都不会给producer端发送消息确认，这就导致consumer正在消费时突然中断会导致数据丢失。如果想让一个consumer消费过程中断时，让其他consumer继续消费，要把这个参数去掉，重写consumer端代码如下：

# -*- coding:utf-8 -*- 
# Author：Brownyangyang
import pika,time

#生成实例
connnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#声明一个管道
channel = connnection.channel()

#声明一个叫number1的队列,如果不声明,consumer如果在producer之前启动会报错
channel.queue_declare(queue='number1')

#ch：管道的内存对象地址
#
def callback(ch,method,properties,body):
    print("---》》》",ch,method,properties)
    time.sleep(30) #模拟收消息过程30s
    print("[x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag) #告诉producer消息收到

##消费消息,如果收到消息，就调用callback函数
channel.basic_consume(callback,
                      queue='number1'
                      )

print("[*] Waiting for messages.CTRL+C to exit.")


#开始监听消息队列
channel.start_consuming()

起两个异常consumer即可测试现象

三、消息持久化 

当我重启前队列是这样的，有一个number1的队列，队列中有一条数据

但是当我重启rabbitmq时,再执行rabbitmqctl list_queues 发现队列没了，队列里的数据也没有了，说明数据现在并不是持久化的

systemctl stop rabbitmq-server.service

systemctl start rabbitmq-server.service

实现持久化需要修改代码如下：

producer端：

# -*- coding:utf-8 -*- 
# Author：Brownyangyang
import pika

#生成一个socket实例
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)

#声明一个管道
channel = connection.channel()

#声明一个叫number1的队列
channel.queue_declare(queue="number1",durable=True)  #durable=True代表持久化队列

#通过管道发送消息，number1是queue名，body是消息主体
channel.basic_publish(exchange='',
                      routing_key="number1",
                      body="hello world",
                      properties=pika.BasicProperties(delivery_mode=2,)  #持久化数据
                      )

print("[x] sent 'hello World!'")
connection.close()

consumer端：

# -*- coding:utf-8 -*- 
# Author：Brownyangyang
import pika,time

#生成实例
connnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#声明一个管道
channel = connnection.channel()

#声明一个叫number1的队列,如果不声明,consumer如果在producer之前启动会报错
channel.queue_declare(queue='number1',durable=True) #durable=True只是把队列持久化

#ch：管道的内存对象地址
#
def callback(ch,method,properties,body):
    print("---》》》",ch,method,properties)
    time.sleep(30) #模拟收消息过程30s
    print("[x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag) #告诉producer消息收到

##消费消息,如果收到消息，就调用callback函数
channel.basic_consume(callback,
                      queue='number1'
                      )

print("[*] Waiting for messages.CTRL+C to exit.")


#开始监听消息队列
channel.start_consuming()

四、消息公平分发

如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，可以在各个消费者端，配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了。

这个在consumer端做改造，如下：

# -*- coding:utf-8 -*- 
# Author：Brownyangyang
import pika,time

#生成实例
connnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#声明一个管道
channel = connnection.channel()

#声明一个叫number1的队列,如果不声明,consumer如果在producer之前启动会报错
channel.queue_declare(queue='number1',durable=True) #durable=True只是把队列持久化

#ch：管道的内存对象地址
#
def callback(ch,method,properties,body):
    print("---》》》",ch,method,properties)
    time.sleep(30) #模拟收消息过程30s
    print("[x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag) #告诉producer消息收到

#如果队列有超过一条数据，producer不向consumer发送数据，可以向其他空闲consumer发送数据，这样保证了负载均衡
channel.basic_qos(prefetch_count=1)

##消费消息,如果收到消息，就调用callback函数
channel.basic_consume(callback,
                      queue='number1'
                      )

print("[*] Waiting for messages.CTRL+C to exit.")


#开始监听消息队列
channel.start_consuming()

五、广播

    有些时候你想让你的消息被所有的Queue收到，类似广播的效果，这时候就要用到exchange了。 producer 发一条消息，所有consumer都能收到（exchange_type='fanout'：所有bind到此exchange的queue都可以接收消息）

producer端：

# -*- coding:utf-8 -*- 
# Author：Brownyangyang
import pika
import sys

connection = pika.BlockingConnection (pika.ConnectionParameters (
    host='localhost'))
channel = connection.channel ( )

#这里是广播，不需要声明queue
channel.exchange_declare (exchange='logs',
                          exchange_type='fanout')

#可以命令行传参，如果不输入默认就是是or后面的消息
message = ' '.join (sys.argv[1:]) or "info: Hello World!"

#routing_key空着，没有queue了
channel.basic_publish (exchange='logs',
                       routing_key='',
                       body=message)
print (" [x] Sent %r" % message)
connection.close ( )

consumer端：

# -*- coding:utf-8 -*- 
# Author：Brownyangyang
import pika

connection = pika.BlockingConnection (pika.ConnectionParameters (
    host='localhost'))
channel = connection.channel ( )

channel.exchange_declare (exchange='logs',
                          exchange_type='fanout')

#queue的对象是result，exclusive，排他唯一
result = channel.queue_declare (exclusive=True)  # 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除

#随机生成一个队列的名子
queue_name = result.method.queue

#绑定队列到logs这个转发器
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 ( )

测试的时候可以启用一个producer：

[root@liyang2 python]# python3 producer3.py
[x] Sent 'info: Hello World!'
[root@liyang2 python]# python3 producer3.py nihao
[x] Sent 'nihao'

再起两个consumer，看是否都能同时收到。

六：有选择的接收消息（exchange_type='direct'）

RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。

producer端：

# -*- coding:utf-8 -*- 
# Author:Brownyangyang
import pika
import sys

connection = pika.BlockingConnection (pika.ConnectionParameters (
    host='localhost'))
channel = connection.channel ( )

channel.exchange_declare (exchange='direct_logs',
                          exchange_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 ( )

consumer端：

# -*- coding:utf-8 -*- 
# Author:Brownyangyang
import pika
import sys

connection = pika.BlockingConnection (pika.ConnectionParameters (
    host='localhost'))
channel = connection.channel ( )

channel.exchange_declare (exchange='direct_logs',
                          exchange_type='direct')

result = channel.queue_declare (exclusive=True)
queue_name = result.method.queue

severities = sys.argv[1:]
if not severities:
    sys.stderr.write ("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 ( )

测试可以起一个producer，三个consumer分别监听info warning error三个级别日志

[root@liyang2 python]# python3 producer4.py info nihao
[x] Sent 'info':'nihao'
[root@liyang2 python]# python3 producer4.py warning nihao
[x] Sent 'warning':'nihao'
[root@liyang2 python]# python3 producer4.py error nihao
[x] Sent 'error':'nihao'

七、更细致的消息过滤（exchange_type='topic'）

     可以根据类似正则表达式模糊匹配消息

python3 consumer5.py "*.error"                             路由后缀为error的信息

python3 consumer5.py "nginx.*"                            路由后缀为nginx开头的信息

python3 consumer5.py "*.error" "*.warning"          路由以error结尾和warning结尾的信息

 python3 consumer5.py "#"                                    路由所有信息，所有消息都消费

producer端： 

# -*- coding:utf-8 -*- 
# Author:Brownyangyang
import pika
import sys

connection = pika.BlockingConnection (pika.ConnectionParameters (
    host='localhost'))
channel = connection.channel ( )

channel.exchange_declare (exchange='topic_logs',
                          exchange_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 ( )

consumer端：

# -*- coding:utf-8 -*- 
# Author:Brownyangyang
import pika
import sys

connection = pika.BlockingConnection (pika.ConnectionParameters (
    host='localhost'))
channel = connection.channel ( )

channel.exchange_declare (exchange='topic_logs',
                          exchange_type='topic')

result = channel.queue_declare (exclusive=True)
queue_name = result.method.queue

binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write ("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 ( )

producer测试命令如下，可以在各自consumer上查看

[root@liyang2 python]# python3 producer5.py 11.error  
[x] Sent '11.error':'Hello World!'
[root@liyang2 python]# python3 producer5.py nginx.error
[x] Sent 'nginx.error':'Hello World!'
[root@liyang2 python]# python3 producer5.py nginx.log
[x] Sent 'nginx.log':'Hello World!'
[root@liyang2 python]# python3 producer5.py 123kkk
[x] Sent '123':'Hello World!'

八、RPC

（未完待续）