Python操作RabbitMQ

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

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

RabbitMQ安装

注意&＃xff1a;service rabbitmq-server start/stop

安装API

使用API操作RabbitMQ

基于Queue实现生产者消费者模型

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import Queue
import threadingmessage &＃61; Queue.Queue(10)def producer(i):while True:message.put(i)def consumer(i):while True:msg &＃61; message.get()for i in range(12):t &＃61; threading.Thread(target&＃61;producer, args&＃61;(i,))t.start()for i in range(10):t &＃61; threading.Thread(target&＃61;consumer, args&＃61;(i,))t.start()

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

1、acknowledgment 消息不丢失

no-ack &＃xff1d; False&＃xff0c;如果消费者遇到情况(its channel is closed, connection is closed, or TCP connection is lost)挂掉了&＃xff0c;那么&＃xff0c;RabbitMQ会重新将该任务添加到队列中。

import pikaconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;10.211.55.4&＃39;))
channel &＃61; connection.channel()channel.queue_declare(queue&＃61;&＃39;hello&＃39;)def callback(ch, method, properties, body):print(" [x] Received %r" % body)import timetime.sleep(10)print &＃39;ok&＃39;ch.basic_ack(delivery_tag &＃61; method.delivery_tag)channel.basic_consume(callback,queue&＃61;&＃39;hello&＃39;,no_ack&＃61;False)print(&＃39; [*] Waiting for messages. To exit press CTRL&＃43;C&＃39;)
channel.start_consuming()

2、durable   消息不丢失

#!/usr/bin/env python
import pikaconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;10.211.55.4&＃39;))
channel &＃61; connection.channel()# make message persistent
channel.queue_declare(queue&＃61;&＃39;hello&＃39;, durable&＃61;True)channel.basic_publish(exchange&＃61;&＃39;&＃39;,routing_key&＃61;&＃39;hello&＃39;,body&＃61;&＃39;Hello World!&＃39;,properties&＃61;pika.BasicProperties(delivery_mode&＃61;2, # make message persistent))
print(" [x] Sent &＃39;Hello World!&＃39;")
connection.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pikaconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;10.211.55.4&＃39;))
channel &＃61; connection.channel()# make message persistent
channel.queue_declare(queue&＃61;&＃39;hello&＃39;, durable&＃61;True)def callback(ch, method, properties, body):print(" [x] Received %r" % body)import timetime.sleep(10)print &＃39;ok&＃39;ch.basic_ack(delivery_tag &＃61; method.delivery_tag)channel.basic_consume(callback,queue&＃61;&＃39;hello&＃39;,no_ack&＃61;False)print(&＃39; [*] Waiting for messages. To exit press CTRL&＃43;C&＃39;)
channel.start_consuming()

3、消息获取顺序

默认消息队列里的数据是按照顺序被消费者拿走&＃xff0c;例如&＃xff1a;消费者1 去队列中获取 奇数 序列的任务&＃xff0c;消费者1去队列中获取 偶数 序列的任务。

channel.basic_qos(prefetch_count&＃61;1) 表示谁来谁取&＃xff0c;不再按照奇偶数排列

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pikaconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;10.211.55.4&＃39;))
channel &＃61; connection.channel()# make message persistent
channel.queue_declare(queue&＃61;&＃39;hello&＃39;)def callback(ch, method, properties, body):print(" [x] Received %r" % body)import timetime.sleep(10)print &＃39;ok&＃39;ch.basic_ack(delivery_tag &＃61; method.delivery_tag)channel.basic_qos(prefetch_count&＃61;1)channel.basic_consume(callback,queue&＃61;&＃39;hello&＃39;,no_ack&＃61;False)print(&＃39; [*] Waiting for messages. To exit press CTRL&＃43;C&＃39;)
channel.start_consuming()

4、发布订阅

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

 exchange type &＃61; fanout

#!/usr/bin/env python
import pika
import sysconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;logs&＃39;,type&＃61;&＃39;fanout&＃39;)message &＃61; &＃39; &＃39;.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange&＃61;&＃39;logs&＃39;,routing_key&＃61;&＃39;&＃39;,body&＃61;message)
print(" [x] Sent %r" % message)
connection.close()

#!/usr/bin/env python
import pikaconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;logs&＃39;,type&＃61;&＃39;fanout&＃39;)result &＃61; channel.queue_declare(exclusive&＃61;True)
queue_name &＃61; result.method.queuechannel.queue_bind(exchange&＃61;&＃39;logs&＃39;,queue&＃61;queue_name)print(&＃39; [*] Waiting for logs. To exit press CTRL&＃43;C&＃39;)def callback(ch, method, properties, body):print(" [x] %r" % body)channel.basic_consume(callback,queue&＃61;queue_name,no_ack&＃61;True)channel.start_consuming()

5、关键字发送

 exchange type &＃61; direct

之前事例&＃xff0c;发送消息时明确指定某个队列并向其中发送消息&＃xff0c;RabbitMQ还支持根据关键字发送&＃xff0c;即&＃xff1a;队列绑定关键字&＃xff0c;发送者将数据根据关键字发送到消息exchange&＃xff0c;exchange根据 关键字 判定应该将数据发送至指定队列。

#!/usr/bin/env python
import pika
import sysconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;direct_logs&＃39;,type&＃61;&＃39;direct&＃39;)result &＃61; channel.queue_declare(exclusive&＃61;True)
queue_name &＃61; result.method.queueseverities &＃61; sys.argv[1:]
if not severities:sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])sys.exit(1)for severity in severities:channel.queue_bind(exchange&＃61;&＃39;direct_logs&＃39;,queue&＃61;queue_name,routing_key&＃61;severity)print(&＃39; [*] Waiting for logs. To exit press CTRL&＃43;C&＃39;)def callback(ch, method, properties, body):print(" [x] %r:%r" % (method.routing_key, body))channel.basic_consume(callback,queue&＃61;queue_name,no_ack&＃61;True)channel.start_consuming()

#!/usr/bin/env python
import pika
import sysconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;direct_logs&＃39;,type&＃61;&＃39;direct&＃39;)severity &＃61; sys.argv[1] if len(sys.argv) > 1 else &＃39;info&＃39;
message &＃61; &＃39; &＃39;.join(sys.argv[2:]) or &＃39;Hello World!&＃39;
channel.basic_publish(exchange&＃61;&＃39;direct_logs&＃39;,routing_key&＃61;severity,body&＃61;message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

6、模糊匹配

 exchange type &＃61; topic

在topic类型下&＃xff0c;可以让队列绑定几个模糊的关键字&＃xff0c;之后发送者将数据发送到exchange&＃xff0c;exchange将传入”路由值“和 ”关键字“进行匹配&＃xff0c;匹配成功&＃xff0c;则将数据发送到指定队列。

• # 表示可以匹配 0 个 或 多个 单词
• *  表示只能匹配 一个 单词

#!/usr/bin/env python
import pika
import sysconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;topic_logs&＃39;,type&＃61;&＃39;topic&＃39;)result &＃61; channel.queue_declare(exclusive&＃61;True)
queue_name &＃61; result.method.queuebinding_keys &＃61; sys.argv[1:]
if not binding_keys:sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])sys.exit(1)for binding_key in binding_keys:channel.queue_bind(exchange&＃61;&＃39;topic_logs&＃39;,queue&＃61;queue_name,routing_key&＃61;binding_key)print(&＃39; [*] Waiting for logs. To exit press CTRL&＃43;C&＃39;)def callback(ch, method, properties, body):print(" [x] %r:%r" % (method.routing_key, body))channel.basic_consume(callback,queue&＃61;queue_name,no_ack&＃61;True)channel.start_consuming()

#!/usr/bin/env python
import pika
import sysconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;topic_logs&＃39;,type&＃61;&＃39;topic&＃39;)routing_key &＃61; sys.argv[1] if len(sys.argv) > 1 else &＃39;anonymous.info&＃39;
message &＃61; &＃39; &＃39;.join(sys.argv[2:]) or &＃39;Hello World!&＃39;
channel.basic_publish(exchange&＃61;&＃39;topic_logs&＃39;,routing_key&＃61;routing_key,body&＃61;message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()

注意&＃xff1a;

sudo rabbitmqctl add_user wupeiqi 123
# 设置用户为administrator角色
sudo rabbitmqctl set_user_tags wupeiqi administrator
# 设置权限
sudo rabbitmqctl set_permissions -p "/" root ".*" ".*" ".*"# 然后重启rabbiMQ服务
sudo /etc/init.d/rabbitmq-server restart# 然后可以使用刚才的用户远程连接rabbitmq server了。------------------------------
credentials &＃61; pika.PlainCredentials("wupeiqi","123")connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.14.47&＃39;,credentials&＃61;credentials))

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
from pika.adapters.blocking_connection import BlockingChannelcredentials &＃61; pika.PlainCredentials("root", "123")conn &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;10.211.55.20&＃39;, credentials&＃61;credentials))
# 超时时间
conn.add_timeout(5, lambda: channel.stop_consuming())channel &＃61; conn.channel()channel.queue_declare(queue&＃61;&＃39;hello&＃39;)def callback(ch, method, properties, body):print(" [x] Received %r" % body)channel.stop_consuming()channel.basic_consume(callback,queue&＃61;&＃39;hello&＃39;,no_ack&＃61;True)print(&＃39; [*] Waiting for messages. To exit press CTRL&＃43;C&＃39;)
channel.start_consuming()

https://www.cnblogs.com/zhangxinqi/p/8460768.html