Rabbitmq队列
一. RabbitMQ队列
#消息中间件 -消息队列 - 异步 提交的任务不须要实时获得结果或回应 #应用 - 减轻服务器压力，提升单位时间处理请求数 - RPC #消息队列 - Q对象 - Redis列表 - RabbitMQ
a. 安装html
#Centos7 安装 #注意/etc/hosts文件 ip和主机名对应 wget https://github.com/rabbitmq/rabbitmq-server/releases/download/rabbitmq_v3_6_10/rabbitmq-server-3.6.10-1.el7.noarch.rpm yum install epel-release -y yum install rabbitmq-server-3.6.10-1.el7.noarch.rpm rabbitmq-plugins enable rabbitmq_management cp /usr/share/doc/rabbitmq-server-3.6.10/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config systemctl restart rabbitmq-server systemctl status rabbitmq-server #建立用户 受权 rabbitmqctl add_user alex alex3714 rabbitmqctl set_permissions -p / alex ".*" ".*" ".*"
b. 建立用户 受权 python
#远程链接rabbitmq server的话,须要配置权限 #建立用户 rabbitmqctl add_user alex alex3714 #同时还要配置权限，容许从外面访问 rabbitmqctl set_permissions -p / alex ".*" ".*" ".*" set_permissions [-p vhost] {user} {conf} {write} {read} vhost The name of the virtual host to which to grant the user access, defaulting to /. user The name of the user to grant access to the specified virtual host. conf A regular expression matching resource names for which the user is granted configure permissions. write A regular expression matching resource names for which the user is granted write permissions. read A regular expression matching resource names for which the user is granted read permissions.
c. python rabbitMQ module 安装mysql
pip install pika or easy_install pika or 源码 https://pypi.python.org/pypi/pika
二. 事例
注意: 通常申明队列(以下代码)只须要在服务端申明,但客户端也能够申明,是防止若是服务端没有启动,客户端先启动后没有队列会报错 此时服务端若是有相同代码,会检查,若是有相同队列就不建立 channel.queue_declare(queue='hello')
a. 消息队列git
#查看队列
# rabbitmqctl list_queues
#客户端再次申明队列是由于客户端要清楚去哪里取数据
channel.queue_declare(queue='hello')
import pika credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('192.168.11.106',credentials=credentials)) #创建socket channel = connection.channel() #建立rabbitmq协议通道 channel.queue_declare(queue='hello') #经过通道生成一个队列 channel.basic_publish(exchange='', routing_key='hello', #队列 body='Hello World!') #内容 print(" [x] Sent 'Hello World!'") connection.close()
import pika credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('192.168.11.106',credentials=credentials)) #创建socket channel = connection.channel() channel.queue_declare(queue='hello') def callback(ch, method, properties, body): print(ch) #上面channel = connection.channel()对象 print(method) #除了服务端自己的数据，还带一些参数 print(properties) #属性 print(body) #byte数据 channel.basic_consume(callback, #监听队列，若是队列中有数据，执行回调函数 queue='hello', no_ack=True) #处理完回调函数不须要回应服务端 print(' Waiting for messages. To exit press CTRL+C') channel.start_consuming() #开始监听
消息持久化之 客户端挂掉，消息还会在服务端。github
1. no_ack=True 模拟客户端中断 观察服务端队列的数据会不会返回(不会)面试
#- 开启一个服务端,两个客户端 #- 服务端向队列中存放一个值,一客户端从队列中取到数据,在睡20秒期间中断,表示出错,它不会报告给服务端 #- 这时队列中为零,另外一客户端也不会取到值 # no_ack=True 表示客户端处理完了不须要向服务端确认消息
import pika credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('192.168.11.106',credentials=credentials)) #创建socket channel = connection.channel() #建立rabbitmq协议通道 channel.queue_declare(queue='hello') #经过通道生成一个队列 channel.basic_publish(exchange='', routing_key='hello', #队列 body='Hello World!') #内容 print(" [x] Sent 'Hello World!'") connection.close()
import pika import time credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('192.168.11.106',credentials=credentials)) #创建socket channel = connection.channel() channel.queue_declare(queue='hello') def callback(ch, method, properties, body): print("received msg...start process",body) time.sleep(10) print("end process...") channel.basic_consume(callback, queue='hello', no_ack=True) print(' Waiting for messages. To exit press CTRL+C') channel.start_consuming()
2. no_ack=False 客户端须要向服务端回应，若是没有回应或抛异常，则服务端队列的数据不会消失，还在队列中。sql
import pika credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('192.168.11.106',credentials=credentials)) #创建socket channel = connection.channel() channel.queue_declare(queue='hello') def callback(ch, method, properties, body): 抛异常 ch.basic_ack(delivery_tag = method.delivery_tag) #客户端回应服务端 channel.basic_consume(callback, #监听队列，若是队列中有数据，执行回调函数 queue='hello', no_ack=False) #处理完回调函数须要回应服务端 print(' Waiting for messages. To exit press CTRL+C') channel.start_consuming() #开始监听
消息持久化 模拟客户端中断 观察服务端队列的数据会不会返回(会) express
#1. 生产者端发消息时，加参数 消息持久化
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
),
#2. 消费者端，消息处理完毕时，发送确认包
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_consume(callback, #取到消息后，调用callback 函数
queue='task1',)
#no_ack=True) #消息处理后，不向rabbit-server确认消息已消费完毕
#- 开启一个服务端,两个客户端
#- 服务端向队列中存放一个值,一客户端从队列中取到数据,在睡20秒期间中断,表示出错,它会报给服务端,服务端队列还有值
#- 这时启动另外一客户端还能够取到值
import pika credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('192.168.11.106',credentials=credentials)) #创建socket channel = connection.channel() #建立rabbitmq协议通道 channel.queue_declare(queue='hello') #经过通道生成一个队列 channel.basic_publish(exchange='', routing_key='hello', #队列 properties=pika.BasicProperties( delivery_mode=2, # make message persistent ), body='Hello World!') #内容 print(" [x] Sent 'Hello World!'") connection.close()
import pika import time credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('192.168.11.106',credentials=credentials)) #创建socket channel = connection.channel() channel.queue_declare(queue='hello') def callback(ch, method, properties, body): print("received msg...start process",body) time.sleep(10) print("end process...") ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_consume(callback, queue='hello', ) print(' Waiting for messages. To exit press CTRL+C') channel.start_consuming()
队列持久化django
#队列持久化
channel.queue_declare(queue='hello',durable=True) # 声明队列持久化
systemctl restart rabbitmq-server #重启服务发现hello队列还在,可是消息不在
rabbitmqctl list_queues
#hello
#队列和消息持久化
channel.queue_declare(queue='hello',durable=True)
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
),
systemctl restart rabbitmq-server #重启服务发现队列和消息都还在
rabbitmqctl list_queues
#hello 6
import pika credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('192.168.11.106',credentials=credentials)) #创建socket channel = connection.channel() #建立rabbitmq协议通道 channel.queue_declare(queue='hello',durable=True) #经过通道生成一个队列 channel.basic_publish(exchange='', routing_key='hello', #队列 properties=pika.BasicProperties( delivery_mode=2, # make message persistent ), body='Hello World!') #内容 print(" [x] Sent 'Hello World!'") connection.close()
b. 发布和订阅 fanout 广播服务器
#服务端： - 不须要申明队列 #客户端： - 每一个客户端都须要申明一个队列,自动设置队列名称,收听广播,当收听完后queue删除 - 把队列绑定到exchange上 #注意：客户端先打开,服务端再打开,客户端会收到消息 #应用： - 微博粉丝在线,博主发消息,粉丝能够收到 #若是服务端先启动向exchange发消息,这时客户端没有启动,没有队列保存数据(exchange不负责保存数据) #这时数据会丢,队列中没有数据 #exchange只负责转发
import pika import sys import time credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('172.16.42.128',credentials=credentials)) #创建socket channel = connection.channel() #建立rabbitmq协议通道 channel.exchange_declare(exchange='logs',type='fanout') message = ' '.join(sys.argv[1:]) or "info: Hello World!" channel.basic_publish(exchange='logs', routing_key='', body=message) print(" Send %r" % message) connection.close()
import pika import time credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('172.16.42.128',credentials=credentials)) #创建socket channel = connection.channel() channel.exchange_declare(exchange='logs', type='fanout') queue_obj = channel.queue_declare(exclusive=True) #随机建立一个队列对象 exclusive=True会在使用此queue的消费者断开后,自动将queue删除 queue_name = queue_obj.method.queue #不指定queue名字,rabbit会随机分配一个名字, channel.queue_bind(exchange='logs',queue=queue_name) #把queue绑定到exchange 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()
e. direct 组播
#客户端一：
- python3 receive1.py info
#客户端二：
- python3 receive1.py error
#客户端三：
- python3 receive1.py warning
#客户端四：
- python3 receive1.py warning error info
#服务端：
- python3 receive1.py warning
import pika import sys import time credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('172.16.42.128',credentials=credentials)) #创建socket channel = connection.channel() #建立rabbitmq协议通道 channel.exchange_declare(exchange='direct_logs',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(" Send %r:%r" % (severity, message)) connection.close()
import pika import time import sys credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('172.16.42.128',credentials=credentials)) #创建socket channel = connection.channel() channel.exchange_declare(exchange='direct_logs',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]\n" % 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()
f. topic 规则传播
#客户端一：
- python3 receive1.py *.django
#客户端二：
- python3 receive1.py mysql.error
#客户端三：
- python3 receive1.py mysql.*
#服务端：
- python3 receive1.py #匹配相应的客户端
import pika import time import sys credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('172.16.42.128',credentials=credentials)) #创建socket channel = connection.channel() channel.exchange_declare(exchange='topic_logs',type='topic') result = channel.queue_declare(exclusive=True) queue_name = result.method.queue binding_keys = sys.argv[1:] if not binding_keys: print(sys.argv[1:]) sys.stderr.write("Usage: %s [binding_key]...\n" % 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()
import pika import sys import time credentials = pika.PlainCredentials("egon","egon123") #受权的帐号 密码 connection = pika.BlockingConnection( pika.ConnectionParameters('172.16.42.128',credentials=credentials)) #创建socket channel = connection.channel() #建立rabbitmq协议通道 channel.exchange_declare(exchange='topic_logs',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))
面试：
第一种 多个客户端，服务端发送数据，多个客户端轮番的过来取数据。至关于一万个任务，10我的帮忙处理数据。
第二种 发布订阅： 广播 组播 关键字广播