标签：none 基本 偶数 read 支持 pen api spl 权限
官方文档看这里：
http://www.rabbitmq.com/getstarted.html
1、什么是rabbitMQ？
RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。
MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。
本质：socket之间的通信。
2、其他的消息队列？
queue：内存中的，一般用于测试，将列表放入内存中。
redis列表：非专业队列，可以做队列。
rabbitMQ：专业队列，消息队列。可以做持久化，可以保证数据的安全。
zeroMQ：基于内存的队列，比rabbitMQ更快，因为是 将数据放到内存。
3、为什么要有消息队列？
1 处理生产者消费者不对等的关系。
可以根据生产者的多少（用户请求的多少），消费者动态的增加或减少。
2 进行数据通信：
- restfull api 传送的通过http协议发送的json格式数据
- webservice传送的通过http协议发送的xml格式数据（在rest api诞生之前做的web网站之间的进行数据交互是通过webservice做的，c#，java居多）
- rpc（远程服务调用），基于socket并使用自己封装的协议进行数据传输，做数据交互。
‘处理者’在消息队列的一端等候任务到来，一旦有人发送请求就会被立即接收，并做处理。在消息队列旁再临时创建一个队列，发送请求之后就在这个队列一端进行等候，并通知处理者将消息处理完之后就放到这个队列中。任务处理完之后处理者将结果放到队列中，等发送者拿到消息之后就删除临时创建的队列。一般在公司内部做交互，通过rpc来进行通信，各个公司使用的协议不一定相同。升级版：给待处理信息写一个函数名，并传参数，让服务端找到这个函数，通过反射执行，获取结果，然后将结果放到队列中。
服务端安装：
安装配置epel源 $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm 安装erlang $ yum -y install erlang 安装RabbitMQ $ yum -y install rabbitmq-server
运行：
方式1：rabbitmq-server （hang住） ctrl+c停止
方式2：systemctl start rabbitmq-server（在后台进程运行）
默认无密码，如果有密码：
1 sudo rabbitmqctl add_user wupeiqi 123
# 设置用户为administrator角色
2 sudo rabbitmqctl set_user_tags peiqi administrator
# 设置权限
3 sudo rabbitmqctl set_permissions -p "/" root ".*" ".*" ".*"
#重启：
4 systemctl restart rabbitmq-server
客户端：
pip3 install pika
三、使用
基于Queue实现生产者消费者模型
import Queue import threading message = Queue.Queue(10) def producer(i): while True: message.put(i) def consumer(i): while True: msg = message.get() for i in range(12): t = threading.Thread(target=producer, args=(i,)) t.start() for i in range(10): t = threading.Thread(target=consumer, args=(i,)) t.start()
对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() # 创建一个队列：s91 channel.queue_declare(queue=‘s91‘) # 向队列s91中发送一个 Hello World! channel.basic_publish(exchange=‘‘,routing_key=‘s91‘,body=‘66‘) connection.close()
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() channel.queue_declare(queue=‘s91‘) def callback(ch, method, properties, body): print(body) channel.basic_consume(callback,queue=‘s91‘,no_ack=True) channel.start_consuming()
################生产者#########
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()
###############消费者#################
#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘))
channel = connection.channel()
channel.queue_declare(queue=‘hello‘)
#进行回调
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
channel.basic_consume(callback,
queue=‘hello‘,
no_ack=True)
print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
channel.start_consuming()
1、acknowledgment 消息不丢失
no-ack ＝ False，如果消费者遇到情况(its channel is closed, connection is closed, or TCP connection is lost)挂掉了，那么，RabbitMQ会重新将该任务添加到队列中。
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() # 创建一个队列：s91 channel.queue_declare(queue=‘s91‘) # 向队列s91中发送一个 Hello World! channel.basic_publish(exchange=‘‘,routing_key=‘s91‘,body=‘66‘) connection.close()
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() # channel.queue_declare(queue=‘s91‘) def callback(ch, method, properties, body): print(body) ## 进行确认 ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_consume(callback,queue=‘s91‘,no_ack=False) channel.start_consuming()
2、durable 消息不丢失（服务端持久化）
就算服务端或客户端挂掉，也没有关系，RabbitMQ会重新将该任务添加到队列中。
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() # make message persistent channel.queue_declare(queue=‘s92‘, durable=True) channel.basic_publish(exchange=‘‘, routing_key=‘s92‘, body=‘Hello World!‘, properties=pika.BasicProperties( delivery_mode=2, # make message persistent )) connection.close()
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() # make message persistent channel.queue_declare(queue=‘s92‘, durable=True) def callback(ch, method, properties, body): print(" [x] Received %r" % body) ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_consume(callback,queue=‘s92‘,no_ack=False) channel.start_consuming()
3、消息获取顺序
默认消息队列里的数据是按照顺序被消费者拿走，例如：消费者1 去队列中获取 奇数 序列的任务，消费者1去队列中获取 偶数 序列的任务。
channel.basic_qos(prefetch_count=1) 表示谁来谁取，不再按照奇偶数排列
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() # make message persistent channel.queue_declare(queue=‘s92‘, durable=True) channel.basic_publish(exchange=‘‘, routing_key=‘s92‘, body=‘Hello World!‘, properties=pika.BasicProperties( delivery_mode=2, # make message persistent )) connection.close()
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() # make message persistent channel.queue_declare(queue=‘s92‘, durable=True) def callback(ch, method, properties, body): print(" [x] Received %r" % body) ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_qos(prefetch_count=1) channel.basic_consume(callback,queue=‘s92‘,no_ack=False) channel.start_consuming()
4、发布订阅
发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。
可有多个订阅者
exchange type = fanout
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘e1‘,exchange_type=‘fanout‘) message = "Hello World!" channel.basic_publish(exchange=‘e1‘,routing_key=‘‘,body=message) connection.close()
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘e1‘,exchange_type=‘fanout‘) # 随机生成对列 名 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue # 让队列和e1绑定 channel.queue_bind(exchange=‘e1‘,queue=queue_name) def callback(ch, method, properties, body): print(" [x] %r" % body) channel.basic_consume(callback,queue=queue_name,no_ack=True) channel.start_consuming()
5、关键字发送
exchange type = direct
之前事例，发送消息时明确指定某个队列并向其中发送消息，RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘e2‘,exchange_type=‘direct‘) message = "Hello World!" channel.basic_publish(exchange=‘e2‘,routing_key=‘error‘,body=message) connection.close()
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘e2‘,exchange_type=‘direct‘) # 随机生成对列名 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue # 让队列和e1绑定 channel.queue_bind(exchange=‘e2‘,queue=queue_name,routing_key=‘info‘) channel.queue_bind(exchange=‘e2‘,queue=queue_name,routing_key=‘error‘) def callback(ch, method, properties, body): print(" [x] %r" % body) channel.basic_consume(callback,queue=queue_name,no_ack=True) channel.start_consuming()
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘e2‘,exchange_type=‘direct‘) # 随机生成对列名 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue # 让队列和e1绑定 channel.queue_bind(exchange=‘e2‘,queue=queue_name,routing_key=‘error‘) def callback(ch, method, properties, body): print(" [x] %r" % body) channel.basic_consume(callback,queue=queue_name,no_ack=True) channel.start_consuming()
6、模糊匹配
exchange type = topic
在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。
|
1
2
3
|
发送者路由值 队列中old.boy.python old.* -- 不匹配old.boy.python old.# -- 匹配 |
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘e3‘,exchange_type=‘topic‘) message = "Hello World!" channel.basic_publish(exchange=‘e3‘,routing_key=‘info.xx.uu‘,body=message) connection.close()
import pika credentials = pika.PlainCredentials("root","123") connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.13.92‘,credentials=credentials)) channel = connection.channel() channel.exchange_declare(exchange=‘e3‘,exchange_type=‘topic‘) # 随机生成对列名 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue # 让队列和e1绑定 channel.queue_bind(exchange=‘e3‘,queue=queue_name,routing_key=‘info.*‘) def callback(ch, method, properties, body): print(" [x] %r" % body) channel.basic_consume(callback,queue=queue_name,no_ack=True) channel.start_consuming()
注意：
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 = pika.PlainCredentials("wupeiqi","123") connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.14.47‘,credentials=credentials)) 复制代码
#!/usr/bin/env python # -*- coding:utf-8 -*- import pika from pika.adapters.blocking_connection import BlockingChannel credentials = pika.PlainCredentials("root", "123") conn = pika.BlockingConnection(pika.ConnectionParameters(host=‘10.211.55.20‘, credentials=credentials)) # 超时时间 conn.add_timeout(5, lambda: channel.stop_consuming()) channel = conn.channel() channel.queue_declare(queue=‘hello‘) def callback(ch, method, properties, body): print(" [x] Received %r" % body) channel.stop_consuming() channel.basic_consume(callback, queue=‘hello‘, no_ack=True) print(‘ [*] Waiting for messages. To exit press CTRL+C‘) channel.start_consuming()
终极归纳：
使用：
a. 普通消息队列
b. 批量向多个队列中发送（订阅者发布者）
c. 根据关键字匹配向队列中发送
d. 模糊匹配向队列中发送
1. exchange的作用？
- exchange和队列进行绑定
- 用户向队列发送数据时，无序再找队列，直接向exchange中发送即可。
2. rabbitmq中有几种exchange？
- fanout，只要绑定就发
- dirct，确定关键字
- topic，模糊匹配
3. 消息持久化和ack
- 服务端（durable）
- 客户端（ack）
4. 超时
5. 消息顺序
6. 其他消息队列？
- queue
- redis列表
- kafka
- zeromq
7. 什么时候用过消息队列？
- 防止消息堆积（消息提醒）
- 订单处理（celery+消息队列）
标签：none 基本 偶数 read 支持 pen api spl 权限
原文地址：https://www.cnblogs.com/amyleell/p/9315196.html
