RabbitMQ（4）
转：&/blog/691265
Direct Exchange&– 处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键
“dog”，则只有被标记为“dog”的消息才被转发，不会转发dog.puppy，也不会转发dog.guard，只会转发dog。&
Java代码&&
Channel&channel&=&connection.createChannel();&&
channel.exchangeDeclare(&exchangeName&,&&direct&);&&&
channel.queueDeclare(&queueName&);&&
channel.queueBind(&queueName&,&&exchangeName&,&&routingKey&);&&
byte[]&messageBodyBytes&=&&hello&world&.getBytes();&&
channel.basicPublish(&exchangeName&,&&routingKey&,&MessageProperties.PERSISTENT_TEXT_PLAIN,&messageBodyBytes);&&
Fanout Exchange&– 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。&
Java代码&&
Channel&channel&=&connection.createChannel();&&
channel.exchangeDeclare(&exchangeName&,&&fanout&);&&&
channel.queueDeclare(&queueName&);&&
channel.queueBind(&queueName&,&&exchangeName&,&&routingKey&);&&
channel.queueDeclare(&queueName1&);&&
channel.queueBind(&queueName1&,&&exchangeName&,&&routingKey1&);&&
byte[]&messageBodyBytes&=&&hello&world&.getBytes();&&
channel.basicPublish(&exchangeName&,&&&,&MessageProperties.PERSISTENT_TEXT_PLAIN,&messageBodyBytes);&&
Topic Exchange&– 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词。因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.*”
只会匹配到“audit.irs”。我在RedHat的朋友做了一张不错的图，来表明topic交换机是如何工作的：&
Java代码&&
Channel&channel&=&connection.createChannel();&&
channel.exchangeDeclare(&exchangeName&,&&topic&);&&&
channel.queueDeclare(&queueName&);&&
channel.queueBind(&queueName&,&&exchangeName&,&&routingKey.*&);&&
byte[]&messageBodyBytes&=&&hello&world&.getBytes();&&
channel.basicPublish(&exchangeName&,&&routingKey.one&,&MessageProperties.PERSISTENT_TEXT_PLAIN,&messageBodyBytes); &
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：14265次
排名：千里之外
转载：47篇
(1)(2)(2)(1)(8)(12)(20)(1)(1)(1)消息队列两个用处：服务间解耦，缓解压力（削峰平谷），以前用过ZMQ、狼厂内部的NMQ，现在接触了java开源的kafka和RabbitMQ。目前先不求甚解，有个大概的认识。
RabbitMQ的安装和入门例子见，挺全的。安装前需要安装erlang，启动方便。
RabbitMQ是一个强壮的消息队列，安装使用都很容易，支持常用的发布/订阅、消息分发功能。
exchange是消息队列的核心，producer只向exchange发送消息，exchange根据routing key向不同的绑定队列发送消息，consumer从各自的队列接收消息。
使用实例(python)：
import pika
import sys
#producer/consumer通用，与rabbitmq建立连接
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs', type='fanout')#声明创建一个名为logs的exchange，类型为fanout，type有fanout/direct/topic/headers
message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs', routing_key='', body=message) #p向某个exchange发送消息，并指出路由key
print " [x] Sent %r" % (message,)
connection.close()
#同上，建立连接、声明exchange
result = channel.queue_declare(exclusive=True)#声明一个队列，名字由系统产生，exclusive指连接断开后删除队列，queue_declare(queue="xx"）可以自己命名队列
queue_name = result.method.queue
channel.queue_bind(exchange='logs', queue=queue_name) #将队列与名为logs的exchange绑定，不指定exchange则与默认的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) #consumer接收消息不需向rabbit发送确认
channel.start_consuming()
消息的负载均衡
round-robin dispatching：多个consumer订阅同一个队列，会按消息到来的顺序逐个分发给consumers，确保消息个数的平衡
fair dispatch：每个consumer处理资源不同（cpu/memory）、每条消息消耗的时间也不同，按照条数平均分配不能充分利用consumer资源。
#解决办法，consumer加如下代码
channel.basic_qos(prefetch_count=1)#每次向consumer发一条消息，直到收到consumer的确认再发下一条
#consumer每次处理完信息后向rabbitmq发送确认。在callback函数结尾处加上ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_consume(callback,queue=queue_name)#默认no_ack=false，rabbitmq会一直等待直到收到确认消息
Broker：简单来说就是消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。
通信协议AMQP(Advanced Message Queuing Protocol)
AMQP模型中，消息在producer中产生，发送到MQ的exchange上，exchange根据配置的路由方式发到相应的Queue上，Queue又将消息发送给consumer，消息从queue到consumer有push和pull两种方式。 消息队列的使用过程大概如下：
客户端连接到消息队列服务器，打开一个channel。
客户端声明一个exchange，并设置相关属性。
客户端声明一个queue，并设置相关属性。
客户端使用routing key，在exchange和queue之间建立好绑定关系。
客户端投递消息到exchange。
exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里。
exchange支持的分发类型
fanout：向所有队列广播消息，不会根据routing-key来分发
direct:&根据routing-key分发不同的消息到相应的订阅队列
channel.basic_public(exchange='logs',routing_key='error',body=message)
#c:订阅routing_key为error的消息，一个队列可以绑定多次，订阅多个routing_key
channel.queue_bind(exchange='logs',queue=queue_name, routing_key='error')
topic:&toipc是direct的升级版，routing-key支持通配符匹配。* 一个词，#0或多个词，如*.orange会接收a.orange/b.orange消息，#.organge会接收a.b.orange/a.orange消息。一个单#，该队列会接收所有的消息，等同于fanout
阅读(...) 评论()MQ全称为Message Queue,&消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。
MQ是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取或者订阅队列中的消息。MQ和JMS类似，但不同的是JMS是SUN JAVA消息中间件服务的一个标准和API定义，而MQ则是遵循了AMQP协议的具体实现和产品。
在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。
　　　Bind:绑定了Queue和Exchange，意即为符合什么样路由规则的消息，将会放置入哪一个消息队列；
　　Broker：简单来说就是消息队列服务器实体。　　Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。　　Queue：消息队列载体，每个消息都会被投入到一个或多个队列。　　Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。　　Routing Key：路由关键字，exchange根据这个关键字进行消息投递。　　vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。　　producer：消息生产者，就是投递消息的程序。　　consumer：消息消费者，就是接受消息的程序。　　channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。
消息队列的使用过程大概如下：
（1）客户端连接到消息队列服务器，打开一个channel。　　（2）客户端声明一个exchange，并设置相关属性。　　（3）客户端声明一个queue，并设置相关属性。　　（4）客户端使用routing key，在exchange和queue之间建立好绑定关系。　　（5）客户端投递消息到exchange。
exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里。
exchange也有几个类型，完全根据key进行投递的叫做Direct交换机，例如，绑定时设置了routing key为&abc&，那么客户端提交的消息，只有设置了key为&abc&的才会投递到队列。对key进行模式匹配后进行投递的叫做Topic交换机，符号&#&匹配一个或多个词，符号&*&匹配正好一个词。例如&abc.#&匹配&abc.def.ghi&，&abc.*&只匹配&abc.def&。还有一种不需要key的，叫做Fanout交换机，它采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列。
RabbitMQ支持消息的持久化，也就是数据写在磁盘上，为了数据安全考虑，我想大多数用户都会选择持久化。消息队列持久化包括3个部分：　　（1）exchange持久化，在声明时指定durable =& 1　　（2）queue持久化，在声明时指定durable =& 1　　（3）消息持久化，在投递时指定delivery_mode =& 2（1是非持久化）
如果exchange和queue都是持久化的，那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化，一个非持久化，就不允许建立绑定。
阅读(...) 评论()}