徐州网站建设技术托管,简述优化搜索引擎的方法,百度网盘优化,app开发者需要更新此app怎么解决介绍RabbitMQ是一个在AMQP基础上完整的#xff0c;可复用的企业消息系统。他遵循Mozilla Public License开源协议。MQ全称为Message Queue,消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信#xff0c;而无需专用…介绍RabbitMQ是一个在AMQP基础上完整的可复用的企业消息系统。他遵循Mozilla Public License开源协议。MQ全称为Message Queue,消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信而不是通过直接调用彼此来通信直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。应用场景RabbitMQ无疑是目前最流行的消息队列之一对各种语言环境的支持也很丰富作为一个.NET developer有必要学习和了解这一工具。消息队列的使用场景大概有3种1、系统集成分布式系统的设计。各种子系统通过消息来对接这种解决方案也逐步发展成一种架构风格即“通过消息传递的架构”。2、当系统中的同步处理方式严重影响了吞吐量比如日志记录。假如需要记录系统中所有的用户行为日志如果通过同步的方式记录日志势必会影响系统的响应速度当我们将日志消息发送到消息队列记录日志的子系统就会通过异步的方式去消费日志消息。3、系统的高可用性比如电商的秒杀场景。当某一时刻应用服务器或数据库服务器收到大量请求将会出现系统宕机。如果能够将请求转发到消息队列再由服务器去消费这些消息将会使得请求变得平稳提高系统的可用性。安装安装RabbitMQ基础环境内核3.10.0-327.el7.x86_64系统版本CentOS Linux release7.2.1511(Core)安装配置epel源#rpm -ivh http://mirrors.neusoft.edu.cn/epel/7/x86_64/e/epel-release-7-7.noarch.rpm安装erlang#yum install erlang下载RabbitMQ 3.6.1#wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.1/rabbitmq-server-3.6.1-1.noarch.rpm安装rabbitmq-server#rpm -ivh rabbitmq-server-3.6.1-1.noarch.rpm生成配置文件#cp /usr/share/doc/rabbitmq-server-3.6.1/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config启动RabbitMQ#rabbitmq-server start安装Python API#pip3 install pikaor#easy_install pikaPython 操作RabbitMQ对于RabbitMQ来说生产和消费不再针对内存里的一个Queue对象而是某台服务器上的RabbitMQ Server实现的消息队列。1.生产者代码#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguopingimportpika########################## 生产者 #########################credentials pika.PlainCredentials(admin, admin)#链接rabbit服务器(localhost是本机如果是其他服务器请修改为ip地址)connection pika.BlockingConnection(pika.ConnectionParameters(192.168.1.103,5672,/,credentials))#创建频道channel connection.channel()#声明消息队列消息将在这个队列中进行传递。如果将消息发送到不存在的队列rabbitmq将会自动清除这些消息。如果队列不存在则创建channel.queue_declare(queuehello)#exchange -- 它使我们能够确切地指定消息应该到哪个队列去。#向队列插入数值 routing_key是队列名 body是要插入的内容channel.basic_publish(exchange,routing_keyhello,bodyHello World!)print(开始队列)#缓冲区已经flush而且消息已经确认发送到了RabbitMQ中关闭链接connection.close()2.消费者代码#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguopingimportpika########################### 消费者 ##########################credentials pika.PlainCredentials(admin, admin)#连接到rabbitmq服务器connection pika.BlockingConnection(pika.ConnectionParameters(192.168.1.103,5672,/,credentials))channelconnection.channel()#声明消息队列消息将在这个队列中进行传递。如果队列不存在则创建channel.queue_declare(queuewzg)#定义一个回调函数来处理这边的回调函数就是将信息打印出来。defcallback(ch, method, properties, body):print([x] Received %r %body)#告诉rabbitmq使用callback来接收信息channel.basic_consume(callback,queuehello,no_ackTrue)#no_ackTrue表示在回调函数中不需要发送确认标识print([*] Waiting for messages. To exit press CTRLC)#开始接收信息并进入阻塞状态队列里有信息才会调用callback进行处理。按ctrlc退出。channel.start_consuming()RabbitMQ持久化1、acknowledgment 消息不丢失的方法生效方法channel.basic_consume(consumer_callback, queue, no_ackFalse, exclusiveFalse, consumer_tagNone, argumentsNone)即no_ackFalse(默认为False即必须有确认标识),在回调函数consumer_callback中未收到确认标识那么RabbitMQ会重新将该任务添加到队列中。生产者代码#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguopingimportpika########################## 生产者 #########################credentials pika.PlainCredentials(admin, admin)#链接rabbit服务器(localhost是本机如果是其他服务器请修改为ip地址)connection pika.BlockingConnection(pika.ConnectionParameters(192.168.1.103,5672,/,credentials))#创建频道channel connection.channel()#声明消息队列消息将在这个队列中进行传递。如果将消息发送到不存在的队列rabbitmq将会自动清除这些消息。如果队列不存在则创建channel.queue_declare(queuehello)#exchange -- 它使我们能够确切地指定消息应该到哪个队列去。#向队列插入数值 routing_key是队列名 body是要插入的内容channel.basic_publish(exchange,routing_keyhello,bodyHello World!)print(开始队列)#缓冲区已经flush而且消息已经确认发送到了RabbitMQ中关闭链接connection.close()消费者代码importpikacredentials pika.PlainCredentials(admin, admin)#链接rabbitconnection pika.BlockingConnection(pika.ConnectionParameters(192.168.1.103,5672,/,credentials))#创建频道channel connection.channel()#如果生产者没有运行创建队列那么消费者创建队列channel.queue_declare(queuehello)defcallback(ch, method, properties, body):print([x] Received %r %body)importtimetime.sleep(10)printokch.basic_ack(delivery_tagmethod.delivery_tag) #主要使用此代码channel.basic_consume(callback,queuehello,no_ackFalse)print([*] Waiting for messages. To exit press CTRLC)channel.start_consuming()2、消息持久化存储(Message durability)虽然有了消息反馈机制但是如果rabbitmq自身挂掉的话那么任务还是会丢失。所以需要将任务持久化存储起来。声明持久化存储channel.queue_declare(queuewzg, durableTrue) #声明队列持久化Ps: 但是这样程序会执行错误因为‘wzg’这个队列已经存在并且是非持久化的rabbitmq不允许使用不同的参数来重新定义存在的队列。因此需要重新定义一个队列channel.queue_declare(queuetest_queue, durableTrue) #声明队列持久化注意如果仅仅是设置了队列的持久化仅队列本身可以在rabbit-server宕机后保留队列中的信息依然会丢失如果想让队列中的信息或者任务保留还需要做以下设置channel.basic_publish(exchange,routing_keytest_queue,bodymessage,propertiespika.BasicProperties(delivery_mode 2, #使消息或任务也持久化存储))消息队列持久化包括3个部分(1)exchange持久化在声明时指定durable 1(2)queue持久化在声明时指定durable 1(3)消息持久化在投递时指定delivery_mode 2(1是非持久化)如果exchange和queue都是持久化的那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化一个非持久化就不允许建立绑定。发布与订阅RabbitMQ的发布与订阅借助于交换机(Exchange)来实现。交换机的工作原理消息发送端先将消息发送给交换机交换机再将消息发送到绑定的消息队列而后每个接收端(consumer)都能从各自的消息队列里接收到信息。Exchange有三种工作模式分别为Fanout, Direct, Topic模式1 Fanout任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上1.可以理解为路由表的模式2.这种模式不需要routing_key(即使指定也是无效的)3.这种模式需要提前将Exchange与Queue进行绑定一个Exchange可以绑定多个Queue一个Queue可以同多个Exchange进行绑定。4.如果接受到消息的Exchange没有与任何Queue绑定则消息会被抛弃。注意这个时候必须先启动消费者即订阅者。因为随机队列是在consumer启动的时候随机生成的并且进行绑定的。producer仅仅是发送至exchange并不直接与随机队列进行通信。生产者代码#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping#rabbitmq 发布者importpikacredentials pika.PlainCredentials(admin, admin)#链接rabbit服务器(localhost是本机如果是其他服务器请修改为ip地址)connection pika.BlockingConnection(pika.ConnectionParameters(192.168.1.103,5672,/,credentials))channelconnection.channel()#定义交换机exchange表示交换机名称type表示类型channel.exchange_declare(exchangelogs_fanout,typefanout)message Hello Python#将消息发送到交换机channel.basic_publish(exchangelogs_fanout, #指定exchangerouting_key, #fanout下不需要配置配置了也不会生效bodymessage)print([x] Sent %r %message)connection.close()消费者代码#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping#rabbitmq 订阅者importpikacredentials pika.PlainCredentials(admin, admin)#链接rabbit服务器(localhost是本机如果是其他服务器请修改为ip地址)connection pika.BlockingConnection(pika.ConnectionParameters(192.168.1.103,5672,/,credentials))channelconnection.channel()#定义交换机进行exchange声明exchange表示交换机名称type表示类型channel.exchange_declare(exchangelogs_fanout,typefanout)#随机创建队列result channel.queue_declare(exclusiveTrue) #exclusiveTrue表示建立临时队列当consumer关闭后该队列就会被删除queue_name result.method.queue#将队列与exchange进行绑定channel.queue_bind(exchangelogs_fanout,queuequeue_name)print([*] Waiting for logs. To exit press CTRLC)defcallback(ch, method, properties, body):print([x] %r %body)#从队列获取信息channel.basic_consume(callback,queuequeue_name,no_ackTrue)channel.start_consuming()模式2 Direct路由键的工作原理每个接收端的消息队列在绑定交换机的时候可以设定相应的路由键。发送端通过交换机发送信息时可以指明路由键 交换机会根据路由键把消息发送到相应的消息队列这样接收端就能接收到消息了。任何发送到Direct Exchange的消息都会被转发到routing_key中指定的Queue1.一般情况可以使用rabbitMQ自带的Exchange”” (该Exchange的名字为空字符串) 也可以自定义Exchange2.这种模式下不需要将Exchange进行任何绑定(bind)操作。当然也可以进行绑定。可以将不同的routing_key与不同的queue进行绑定不同的queue与不同exchange进行绑定3.消息传递时需要一个“routing_key”4.如果消息中中不存在routing_key中绑定的队列名则该消息会被抛弃。如果一个exchange 声明为direct并且bind中指定了routing_key,那么发送消息时需要同时指明该exchange和routing_key.消费者代码#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping#消费者importpikacredentials pika.PlainCredentials(admin, admin)#链接rabbit服务器(localhost是本机如果是其他服务器请修改为ip地址)connection pika.BlockingConnection(pika.ConnectionParameters(192.168.1.103,5672,/,credentials))channelconnection.channel()#定义exchange和类型channel.exchange_declare(exchangedirect_test,typedirect)#生成随机队列result channel.queue_declare(exclusiveTrue)queue_nameresult.method.queueseverities [error, ]#将随机队列与routing_key关键字以及exchange进行绑定for severity inseverities:channel.queue_bind(exchangedirect_test,queuequeue_name,routing_keyseverity)print([*] Waiting for logs. To exit press CTRLC)defcallback(ch, method, properties, body):print([x] %r:%r %(method.routing_key, body))#接收消息channel.basic_consume(callback,queuequeue_name,no_ackTrue)channel.start_consuming()生产者#!/usr/bin/env python#-*- coding: utf-8 -*-#auth : pangguoping#发布者importpikacredentials pika.PlainCredentials(admin, admin)#链接rabbit服务器(localhost是本机如果是其他服务器请修改为ip地址)connection pika.BlockingConnection(pika.ConnectionParameters(192.168.1.103,5672,/,credentials))channelconnection.channel()#定义交换机名称及类型channel.exchange_declare(exchangedirect_test,typedirect)severity infomessage 123#发布消息至交换机direct_test且发布的消息携带的关键字routing_key是infochannel.basic_publish(exchangedirect_test,routing_keyseverity,bodymessage)print([x] Sent %r:%r %(severity, message))connection.close()当接收端正在运行时可以使用rabbitmqctl list_bindings来查看绑定情况。模式3 Topic路由键模糊匹配其实是路由键(routing_key)的扩展就是可以使用正则表达式和常用的正则表示式不同这里的话“#”表示所有、全部的意思“*”只匹配到一个词。任何发送到Topic Exchange的消息都会被转发到所有关心routing_key中指定话题的Queue上1.这种模式较为复杂简单来说就是每个队列都有其关心的主题所有的消息都带有一个“标题”(routing_key)Exchange会将消息转发到所有关注主题能与 routing_key模糊匹配的队列。2.这种模式需要routing_key也许要提前绑定Exchange与Queue。3.在进行绑定时要提供一个该队列关心的主题如“#.log.#”表示该队列关心所有涉及log的消息(一个routing_key为”MQ.log.error”的消息会被转发到该队列)。4.“#”表示0个或若干个关键字“*”表示一个关键字。如“log.*”能与“log.warn”匹配无法与“log.warn.timeout”匹配但是“log.#”能与上述两者匹配。5.同样如果Exchange没有发现能够与routing_key匹配的Queue则会抛弃此消息。具体代码这里不在多余写参照第二种模式的就可以唯一变动的地方就是exchange type的声明以及进行绑定和发送的时候routing_key使用正则模式即可。至此利用Python操作RabbitMQ以及一些简单用法在这里介绍完毕。参考http://www.cnblogs.com/jishuweiwang
