RabbitMQ基础整理

Posted sunjiafeng105

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ基础整理相关的知识，希望对你有一定的参考价值。

引言

你是否遇到过两个（多个）系统间需要通过定时任务来同步某些数据？你是否在为异构系统的不同进程间相互调用、通讯的问题而苦恼、挣扎？如果是，那么恭喜你，消息服务让你可以很轻松地解决这些问题。
消息服务擅长于解决多系统、异构系统间的数据交换（消息通知/通讯）问题，你也可以把它用于系统间服务的相互调用（RPC）。本文将要介绍的RabbitMQ就是当前最主流的消息中间件之一。

RabbitMQ简介

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。
AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、php、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。
下面将重点介绍RabbitMQ中的一些基础概念，了解了这些概念，是使用好RabbitMQ的基础。

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。ConnectionFactory为Connection的制造工厂。
Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

Queue

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。
技术图片

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。

技术图片

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。
这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

Message durability

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

Prefetch count

前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

技术图片

Exchange

在上一节我们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

技术图片

Exchange是按照什么逻辑将消息路由到Queue的？这个将在Binding一节介绍。
RabbitMQ中的Exchange有四种类型，不同的类型有着不同的路由策略，这将在Exchange Types一节介绍。

routing key

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。
在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。
RabbitMQ为routing key设定的长度限制为255 bytes。

Binding

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。
技术图片

 

Binding key

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。
在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。
binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

Exchange Types

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

fanout

fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。
技术图片

 

上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

direct

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。

技术图片

 

以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

topic

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

  • routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
  • binding key与routing key一样也是句点号“. ”分隔的字符串
  • binding key中可以存在两种特殊字符“*”与“#”，用于做模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）

 

技术图片

 

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。

headers

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。
在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。
该类型的Exchange没有用到过（不过也应该很有用武之地），所以不做介绍。

RPC

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。
但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

技术图片

 


RabbitMQ中实现RPC的机制是：

  • 客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14中properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
  • 服务器端收到消息并处理
  • 服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性
  • 客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理

总结

本文介绍了RabbitMQ中个人认为最重要的概念，充分利用RabbitMQ提供的这些功能就可以处理我们绝大部分的异步业务了。
本篇的基本概念可能很难理解并消化，结合实际的应用代码应该会比较容易吸收。所以接下来要写的文章例中会包含实际的业务应用场景分析，为什么使用RabbitMQ来实现，如何用RabbitMQ来实现。

 

 

http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/tutorials_with_python/[6]RPC/#_5 

 

 

 

http://blog.csdn.net/anzhsoft/article/details/19563091

 

一个客户端可以和多个exchange相连

技术图片

consumer指定的时候只要指定QueueName就OK了    

 

这个系统架构图版权属于sunjun041640。

    RabbitMQ Server： 也叫broker server，它不是运送食物的卡车，而是一种传输服务。原话是RabbitMQisn’t a food truck, it’s a delivery service. 他的角色就是维护一条从Producer到Consumer的路线，保证数据能够按照指定的方式进行传输。但是这个保证也不是100%的保证，但是对于普通的应用来说这已经足够了。当然对于商业系统来说，可以再做一层数据一致性的guard，就可以彻底保证系统的一致性了。

    Client A & B： 也叫Producer，数据的发送方。createmessages and publish (send) them to a broker server (RabbitMQ).一个Message有两个部分：payload（有效载荷）和label（标签）。payload顾名思义就是传输的数据。label是exchange的名字或者说是一个tag，它描述了payload，而且RabbitMQ也是通过这个label来决定把这个Message发给哪个Consumer。AMQP仅仅描述了label，而RabbitMQ决定了如何使用这个label的规则。

    Client 1，2，3：也叫Consumer，数据的接收方。Consumersattach to a broker server (RabbitMQ) and subscribe to a queue。把queue比作是一个有名字的邮箱。当有Message到达某个邮箱后，RabbitMQ把它发送给它的某个订阅者即Consumer。当然可能会把同一个Message发送给很多的Consumer。在这个Message中，只有payload，label已经被删掉了。对于Consumer来说，它是不知道谁发送的这个信息的。就是协议本身不支持。但是当然了如果Producer发送的payload包含了Producer的信息就另当别论了。

     对于一个数据从Producer到Consumer的正确传递，还有三个概念需要明确：exchanges, queues and bindings。

        Exchanges are where producers publish their messages.

        Queuesare where the messages end up and are received by consumers

        Bindings are how the messages get routed from the exchange to particular queues.

   还有几个概念是上述图中没有标明的，那就是Connection（连接），Channel（通道，频道）。

 

   Connection： 就是一个TCP的连接。Producer和Consumer都是通过TCP连接到RabbitMQ Server的。以后我们可以看到，程序的起始处就是建立这个TCP连接。

   Channels： 虚拟连接。它建立在上述的TCP连接中。数据流动都是在Channel中进行的。也就是说，一般情况是程序起始建立TCP连接，第二步就是建立这个Channel。

    那么，为什么使用Channel，而不是直接使用TCP连接？

    对于OS来说，建立和关闭TCP连接是有代价的，频繁的建立关闭TCP连接对于系统的性能有很大的影响，而且TCP的连接数也有限制，这也限制了系统处理高并发的能力。但是，在TCP连接中建立Channel是没有上述代价的。对于Producer或者Consumer来说，可以并发的使用多个Channel进行Publish或者Receive。有实验表明，1s的数据可以Publish10K的数据包。当然对于不同的硬件环境，不同的数据包大小这个数据肯定不一样，但是我只想说明，对于普通的Consumer或者Producer来说，这已经足够了。如果不够用，你考虑的应该是如何细化split你的设计。

 

 

对于API的使用可以参考官方的java文档，（里面的思想无论是java、cpp还是python都相似）

https://www.rabbitmq.com/releases/rabbitmq-java-client/v3.1.1/rabbitmq-java-client-javadoc-3.1.1/com/rabbitmq/client/Channel.html#basicConsume(java.lang.String, boolean, java.lang.String, com.rabbitmq.client.Consumer)

 

 

http://m.oschina.net/blog/186202

从AMQP协议可以看出，MessageQueue、Exchange和Binding构成了AMQP协议的核心，下面我们就围绕这三个主要组件    从应用使用的角度全面的介绍如何利用Rabbit MQ构建消息队列以及使用过程中的注意事项。

技术图片

 

 

  • 1. 声明MessageQueue

      在Rabbit MQ中，无论是生产者发送消息还是消费者接受消息，都首先需要声明一个MessageQueue。这就存在一个问题，是生产者声明还是消费者声明呢？要解决这个问题，首先需要明确：

a)消费者是无法订阅或者获取不存在的MessageQueue中信息。

b)消息被Exchange接受以后，如果没有匹配的Queue，则会被丢弃。

在明白了上述两点以后，就容易理解如果是消费者去声明Queue，就有可能会出现在声明Queue之前，生产者已发送的消息被丢弃的隐患。如果应用能够通过消息重发的机制允许消息丢失，则使用此方案没有任何问题。但是如果不能接受该方案，这就需要无论是生产者还是消费者，在发送或者接受消息前，都需要去尝试建立消息队列。这里有一点需要明确，如果客户端尝试建立一个已经存在的消息队列，Rabbit MQ不会做任何事情，并返回客户端建立成功的。

       如果一个消费者在一个信道中正在监听某一个队列的消息，Rabbit MQ是不允许该消费者在同一个channel去声明其他队列的。Rabbit MQ中，可以通过queue.declare命令声明一个队列，可以设置该队列以下属性:

a) Exclusive：排他队列，如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除。这里需要注意三点：其一，排他队列是基于连接可见的，同一连接的不同信道是可以同时访问同一个连接创建的排他队列的。其二，“首次”，如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同。其三，即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除的。这种队列适用于只限于一个客户端发送读取消息的应用场景。

b)   Auto-delete:自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。

 c)   Durable:持久化，这个会在后面作为专门一个章节讨论。

d)  其他选项，例如如果用户仅仅想查询某一个队列是否已存在，如果不存在，不想建立该队列，仍然可以调用queue.declare，只不过需要将参数passive设为true，传给queue.declare，如果该队列已存在，则会返回true；如果不存在，则会返回Error，但是不会创建新的队列。

  • 2. 生产者发送消息

        在AMQP模型中，Exchange是接受生产者消息并将消息路由到消息队列的关键组件。ExchangeType和Binding决定了消息的路由规则。所以生产者想要发送消息，首先必须要声明一个Exchange和该Exchange对应的Binding。可以通过 ExchangeDeclare和BindingDeclare完成。在Rabbit MQ中，声明一个Exchange需要三个参数：ExchangeName，ExchangeType和Durable。ExchangeName是该Exchange的名字，该属性在创建Binding和生产者通过publish推送消息时需要指定。ExchangeType，指Exchange的类型，在RabbitMQ中，有三种类型的Exchange：direct ，fanout和topic，不同的Exchange会表现出不同路由行为。Durable是该Exchange的持久化属性，这个会在消息持久化章节讨论。声明一个Binding需要提供一个QueueName，ExchangeName和BindingKey。下面我们就分析一下不同的ExchangeType表现出的不同路由规则。

        生产者在发送消息时，都需要指定一个RoutingKey和Exchange，Exchange在接到该RoutingKey以后，会判断该ExchangeType:

                         a) 如果是Direct类型，则会将消息中的RoutingKey与该Exchange关联的所有Binding中的BindingKey进行比较，如果相等，则发送到该Binding对应的Queue中。

技术图片 

                  b)   如果是  Fanout  类型，则会将消息发送给所有与该  Exchange  定义过  Binding  的所有  Queues  中去，其实是一种广播行为。
           

技术图片 

        c)如果是Topic类型，则会按照正则表达式，对RoutingKey与BindingKey进行匹配，如果匹配成功，则发送到对应的Queue中。

             技术图片

  • 3. 消费者订阅消息    

    在RabbitMQ中消费者有2种方式获取队列中的消息:

       a)  一种是通过basic.consume命令，订阅某一个队列中的消息,channel会自动在处理完上一条消息之后，接收下一条消息。（同一个channel消息处理是串行的）。除非关闭channel或者取消订阅，否则客户端将会一直接收队列的消息。

             b)  另外一种方式是通过basic.get命令主动获取队列中的消息，但是绝对不可以通过循环调用basic.get来代替basic.consume，这是因为basic.get RabbitMQ在实际执行的时候，是首先consume某一个队列，然后检索第一条消息，然后再取消订阅。如果是高吞吐率的消费者，最好还是建议使用basic.consume。

      如果有多个消费者同时订阅同一个队列的话，RabbitMQ是采用循环的方式分发消息的，每一条消息只能被一个订阅者接收。例如，有队列Queue，其中ClientA和ClientB都Consume了该队列，MessageA到达队列后，被分派到ClientA，ClientA回复服务器收到响应，服务器删除MessageA；再有一条消息MessageB抵达队列，服务器根据“循环推送”原则，将消息会发给ClientB，然后收到ClientB的确认后，删除MessageB；等到再下一条消息时，服务器会再将消息发送给ClientA。

       这里我们可以看出，消费者再接到消息以后，都需要给服务器发送一条确认命令，这个即可以在handleDelivery里显示的调用basic.ack实现，也可以在Consume某个队列的时候，设置autoACK属性为true实现。这个ACK仅仅是通知服务器可以安全的删除该消息，而不是通知生产者，与RPC不同。 如果消费者在接到消息以后还没来得及返回ACK就断开了连接，消息服务器会重传该消息给下一个订阅者，如果没有订阅者就会存储该消息。

        既然RabbitMQ提供了ACK某一个消息的命令，当然也提供了Reject某一个消息的命令。当客户端发生错误，调用basic.reject命令拒绝某一个消息时，可以设置一个requeue的属性，如果为true，则消息服务器会重传该消息给下一个订阅者；如果为false，则会直接删除该消息。当然，也可以通过ack，让消息服务器直接删除该消息并且不会重传。

  • 4. 持久化：

        Rabbit MQ默认是不持久队列、Exchange、Binding以及队列中的消息的，这意味着一旦消息服务器重启，所有已声明的队列，Exchange，Binding以及队列中的消息都会丢失。通过设置Exchange和MessageQueue的durable属性为true，可以使得队列和Exchange持久化，但是这还不能使得队列中的消息持久化，这需要生产者在发送消息的时候，将delivery mode设置为2，只有这3个全部设置完成后，才能保证服务器重启不会对现有的队列造成影响。这里需要注意的是，只有durable为true的Exchange和durable为ture的Queues才能绑定，否则在绑定时，RabbitMQ都会抛错的。持久化会对RabbitMQ的性能造成比较大的影响，可能会下降10倍不止。

  • 5. 事务：

     对事务的支持是AMQP协议的一个重要特性。假设当生产者将一个持久化消息发送给服务器时，因为consume命令本身没有任何Response返回，所以即使服务器崩溃，没有持久化该消息，生产者也无法获知该消息已经丢失。如果此时使用事务，即通过txSelect()开启一个事务，然后发送消息给服务器，然后通过txCommit()提交该事务，即可以保证，如果txCommit()提交了，则该消息一定会持久化，如果txCommit()还未提交即服务器崩溃，则该消息不会服务器就收。当然Rabbit MQ也提供了txRollback()命令用于回滚某一个事务。

  • 6. Confirm机制：

      使用事务固然可以保证只有提交的事务，才会被服务器执行。但是这样同时也将客户端与消息服务器同步起来，这背离了消息队列解耦的本质。Rabbit MQ提供了一个更加轻量级的机制来保证生产者可以感知服务器消息是否已被路由到正确的队列中——Confirm。如果设置channel为confirm状态，则通过该channel发送的消息都会被分配一个唯一的ID，然后一旦该消息被正确的路由到匹配的队列中后，服务器会返回给生产者一个Confirm，该Confirm包含该消息的ID，这样生产者就会知道该消息已被正确分发。对于持久化消息，只有该消息被持久化后，才会返回Confirm。Confirm机制的最大优点在于异步，生产者在发送消息以后，即可继续执行其他任务。而服务器返回Confirm后，会触发生产者的回调函数，生产者在回调函数中处理Confirm信息。如果消息服务器发生异常，导致该消息丢失，会返回给生产者一个nack，表示消息已经丢失，这样生产者就可以通过重发消息，保证消息不丢失。Confirm机制在性能上要比事务优越很多。但是Confirm机制，无法进行回滚，就是一旦服务器崩溃，生产者无法得到Confirm信息，生产者其实本身也不知道该消息吃否已经被持久化，只有继续重发来保证消息不丢失，但是如果原先已经持久化的消息，并不会被回滚，这样队列中就会存在两条相同的消息，系统需要支持去重。

 

  • 其他：

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。

 

 

 

Linux下安装RabbitMQ 

标签： rabbitmq消息队列
2015-01-24 23:12 777人阅读 评论(0) 收藏 举报

1. 安装erlang：

下载所需的源码：wget http://erlang.org/download/otp_src_R13B04.tar.gz wget http://erlang.org/download/otp_src_R13B04.tar.gz

然后./configure && make && make install

 

注：在configure之后发现有以下提示信息：

(1) odbc : ODBC library – link check failed

需要yum -y install unixODBC unixODBC-devel

(2)“ wxWidgets not found, wx will NOT be usable”及“fop is missing.”这两个可以忽略。

 

 

安装完毕，在命令行键入erl，将会出现如下命令行：

Erlang R13B04 (erts-5.7.5) [source] [smp:2:2] [rq:2][async-threads:0] [hipe] [kernel-poll:false]

Eshell V5.7.5 

(abort with ^G)
1>

 

2.安装RabbitMQ

RabbitMQ的安装有很多版本，我们使用Generic Unix版本。

在http://www.rabbitmq.com/install-generic-unix.html中下载http://www.rabbitmq.com/releases/rabbitmq-server/v3.4.3/rabbitmq-server-generic-unix-3.4.3.tar.gz并解压到Linux中的任意目录(/mysoftware/rabbitmq_server-3.4.3/)，解压后即可使用。

 

3.RabbitMQ开启与关闭

进入rabbitmq安装目录下的sbin目录，分别执行以下命令：

[plain] view plaincopyprint?
  1. 开启RabbitMQ服务：.  
  2. ./rabbitmq-server -detached (-detached为可选参数，表示后台开启)  
  3.   
  4. 开启RabbitMQ管理工具，通过浏览器访问http://localhost:15672使用：.  
  5. ./rabbitmq-plugins enable rabbitmq_management （3.3.1版本以后默认不允许用guest远程管理。可以手动创建管理员帐号来远程管理，具体参见4.4）  
  6.   
  7. 关闭RabbitMQ服务：  
  8. ./rabbitmqctl stop  
开启RabbitMQ服务：.
./rabbitmq-server -detached (-detached为可选参数，表示后台开启)

开启RabbitMQ管理工具，通过浏览器访问http://localhost:15672使用：.
./rabbitmq-plugins enable rabbitmq_management （3.3.1版本以后默认不允许用guest远程管理。可以手动创建管理员帐号来远程管理，具体参见4.4）

关闭RabbitMQ服务：
./rabbitmqctl stop
 
 
注意：如果是登录远程管理界面(比如物理机访问虚拟机的管理界面时)，除了关闭防火墙开通相应允许的端口以外，也不能用guest/guest访问，而应该新创建用户并赋予管理员权限才可访问
增加用户：rabbitmqctl add_user ddd ddd
设置管理员权限： rabbitmqctl set_user_tags diego administrator 这样才可以访问web页面提供的管理页面
./rabbitmqctl  set_permissions -p / diego ‘.*‘ ‘.*‘ ‘.*‘  给用户diego加上有访问虚拟空间/的权限
 

4.设置RabbitMQ管理权限

 

Rabbitmq服务器的主要通过rabbitmqctl和rabbimq-plugins两个工具来管理，以下是一些常用功能。

1）. 服务器启动与关闭

      启动: rabbitmq-server –detached

      关闭:rabbitmqctl stop

      若单机有多个实例，则在rabbitmqctlh后加–n 指定名称

2）. 插件管理

      开启某个插件：rabbitmq-pluginsenable xxx

      关闭某个插件：rabbitmq-pluginsdisablexxx

      注意：重启服务器后生效。

3）.virtual_host管理

      新建virtual_host: rabbitmqctladd_vhost  xxx

      撤销virtual_host:rabbitmqctl  delete_vhost xxx

4）. 用户管理

      新建用户：rabbitmqctl add_user xxx pwd

      删除用户:   rabbitmqctl delete_user xxx

      改密码: rabbimqctl change_password {username} {newpassword}

      设置用户角色：rabbitmqctl set_user_tags {username} {tag ...}

              Tag可以为 administrator,monitoring, management

5）. 权限管理

      权限设置：set_permissions [-pvhostpath] {user} {conf} {write} {read}

               Vhostpath

               Vhost路径

               user

     用户名

              Conf

      一个正则表达式match哪些配置资源能够被该用户访问。

              Write

      一个正则表达式match哪些配置资源能够被该用户读。

               Read

      一个正则表达式match哪些配置资源能够被该用户访问。

6）. 获取服务器状态信息

       服务器状态：rabbitmqctl status

       队列信息：rabbitmqctl list_queues[-p vhostpath] [queueinfoitem ...]

                Queueinfoitem可以为：name，durable，auto_delete，arguments，messages_ready，

                messages_unacknowledged，messages，consumers，memory

       Exchange信息：rabbitmqctllist_exchanges[-p vhostpath] [exchangeinfoitem ...]

                 Exchangeinfoitem有：name，type，durable，auto_delete，internal，arguments.

       Binding信息：rabbitmqctllist_bindings[-p vhostpath] [bindinginfoitem ...]       

                 Bindinginfoitem有：source_name，source_kind，destination_name，destination_kind，routing_key，arguments

       Connection信息：rabbitmqctllist_connections [connectioninfoitem ...]

       Connectioninfoitem有：recv_oct，recv_cnt，send_oct，send_cnt，send_pend等。

       Channel信息：rabbitmqctl  list_channels[channelinfoitem ...]

      Channelinfoitem有consumer_count，messages_unacknowledged，messages_uncommitted，acks_uncommitted，messages_unconfirmed，prefetch_count，client_flow_blocked

 

 

 

标签： rabbitmq消息队列
2015-01-24 23:12 777人阅读 评论(0) 收藏 举报

1. 安装erlang：

下载所需的源码：wget http://erlang.org/download/otp_src_R13B04.tar.gz wget http://erlang.org/download/otp_src_R13B04.tar.gz

然后./configure && make && make install

 

注：在configure之后发现有以下提示信息：

(1) odbc : ODBC library – link check failed

需要yum -y install unixODBC unixODBC-devel

(2)“ wxWidgets not found, wx will NOT be usable”及“fop is missing.”这两个可以忽略。

 

 

安装完毕，在命令行键入erl，将会出现如下命令行：

Erlang R13B04 (erts-5.7.5) [source] [smp:2:2] [rq:2][async-threads:0] [hipe] [kernel-poll:false]

Eshell V5.7.5 

(abort with ^G)
1>

 

2.安装RabbitMQ

RabbitMQ的安装有很多版本，我们使用Generic Unix版本。

在http://www.rabbitmq.com/install-generic-unix.html中下载http://www.rabbitmq.com/releases/rabbitmq-server/v3.4.3/rabbitmq-server-generic-unix-3.4.3.tar.gz并解压到Linux中的任意目录(/mysoftware/rabbitmq_server-3.4.3/)，解压后即可使用。

 

3.RabbitMQ开启与关闭

进入rabbitmq安装目录下的sbin目录，分别执行以下命令：

[plain] view plaincopyprint?
  1. 开启RabbitMQ服务：.  
  2. ./rabbitmq-server -detached (-detached为可选参数，表示后台开启)  
  3.   
  4. 开启RabbitMQ管理工具，通过浏览器访问http://localhost:15672使用：.  
  5. ./rabbitmq-plugins enable rabbitmq_management （3.3.1版本以后默认不允许用guest远程管理。可以手动创建管理员帐号来远程管理，具体参见4.4）  
  6.   
  7. 关闭RabbitMQ服务：  
  8. ./rabbitmqctl stop  
开启RabbitMQ服务：.
./rabbitmq-server -detached (-detached为可选参数，表示后台开启)

开启RabbitMQ管理工具，通过浏览器访问http://localhost:15672使用：.
./rabbitmq-plugins enable rabbitmq_management （3.3.1版本以后默认不允许用guest远程管理。可以手动创建管理员帐号来远程管理，具体参见4.4）

关闭RabbitMQ服务：
./rabbitmqctl stop
 
 
注意：如果是登录远程管理界面(比如物理机访问虚拟机的管理界面时)，除了关闭防火墙开通相应允许的端口以外，也不能用guest/guest访问，而应该新创建用户并赋予管理员权限才可访问
增加用户：rabbitmqctl add_user ddd ddd
设置管理员权限： rabbitmqctl set_user_tags diego administrator 这样才可以访问web页面提供的管理页面
./rabbitmqctl  set_permissions -p / diego ‘.*‘ ‘.*‘ ‘.*‘  给用户diego加上有访问虚拟空间/的权限
 

4.设置RabbitMQ管理权限

 

Rabbitmq服务器的主要通过rabbitmqctl和rabbimq-plugins两个工具来管理，以下是一些常用功能。

1）. 服务器启动与关闭

      启动: rabbitmq-server –detached

      关闭:rabbitmqctl stop

      若单机有多个实例，则在rabbitmqctlh后加–n 指定名称

2）. 插件管理

      开启某个插件：rabbitmq-pluginsenable xxx

      关闭某个插件：rabbitmq-pluginsdisablexxx

      注意：重启服务器后生效。

3）.virtual_host管理

      新建virtual_host: rabbitmqctladd_vhost  xxx

      撤销virtual_host:rabbitmqctl  delete_vhost xxx

4）. 用户管理

      新建用户：rabbitmqctl add_user xxx pwd

      删除用户:   rabbitmqctl delete_user xxx

      改密码: rabbimqctl change_password {username} {newpassword}

      设置用户角色：rabbitmqctl set_user_tags {username} {tag ...}

              Tag可以为 administrator,monitoring, management

5）. 权限管理

      权限设置：set_permissions [-pvhostpath] {user} {conf} {write} {read}

               Vhostpath

               Vhost路径

               user

     用户名

              Conf

      一个正则表达式match哪些配置资源能够被该用户访问。

              Write

      一个正则表达式match哪些配置资源能够被该用户读。

               Read

      一个正则表达式match哪些配置资源能够被该用户访问。

6）. 获取服务器状态信息

       服务器状态：rabbitmqctl status

       队列信息：rabbitmqctl list_queues[-p vhostpath] [queueinfoitem ...]

                Queueinfoitem可以为：name，durable，auto_delete，arguments，messages_ready，

                messages_unacknowledged，messages，consumers，memory

       Exchange信息：rabbitmqctllist_exchanges[-p vhostpath] [exchangeinfoitem ...]

                 Exchangeinfoitem有：name，type，durable，auto_delete，internal，arguments.

       Binding信息：rabbitmqctllist_bindings[-p vhostpath] [bindinginfoitem ...]       

                 Bindinginfoitem有：source_name，source_kind，destination_name，destination_kind，routing_key，arguments

       Connection信息：rabbitmqctllist_connections [connectioninfoitem ...]

       Connectioninfoitem有：recv_oct，recv_cnt，send_oct，send_cnt，send_pend等。

       Channel信息：rabbitmqctl  list_channels[channelinfoitem ...]

      Channelinfoitem有consumer_count，messages_unacknowledged，messages_uncommitted，acks_uncommitted，messages_unconfirmed，prefetch_count，client_flow_blocked

 

 

 

rabbitmq 实现原理

AMQP(高级消息队列协议 Advanced Message Queue Protocol)

AMQP当中有四个概念非常重要: 虚拟主机（virtual host），交换机（exchange），队列（queue）和绑定（binding）。一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，RabbitMQ当中，用户只能在虚拟主机的粒度进行权限控制。因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创 建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。

Producer 要产生消息必须要创建一个 Exchange ，Exchange 用于转发消息，但是它不会做存储，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息，当然如果消息总是发送过去就被直接丢弃那就没有什么意思了，一个 Consumer 想要接受消息的话，就要创建一个 Queue ，并把这个 Queue bind 到指定的 Exchange 上，然后 Exchange 会把消息转发到 Queue 那里，Queue 会负责存储消息，Consumer 可以通过主动 Pop 或者是 Subscribe 之后被动回调的方式来从 Queue 中取得消息。

 

Exchange，Queue，RoutingKey

 

技术图片

蓝色-- Client（相对于Rabbitmq Server来说）

绿色--Exchange

红色—Queue

     - 交换器（Exchange），它是发送消息的实体。

     - 队列（Queue），这是接收消息的实体。

     - 绑定器（Bind），将交换器和队列连接起来，并且封装消息的路由信息。

 

Exchange指向Queue的黑色线—RoutingKey，可以将它简单的理解为一条连接Exchange和Queue的路线

 

Exchange和Queue都需要通过channel来进行定义，而RoutingKey则只需要在binding时取个名字就行了。

 

这一块的理解是不正确的，

Exchange Queue RoutingKey关系说明：

 

Exchange Name

Queue Name

Routing Key

 

test.queue     

test.queue

 

test.queue2    

test.queue2

test.exchange  

test.queue     

test.routingkey

test.exchange  

test.queue2    

test.routingkey

test.exchange  

test.queue     

test.routingkey2

test.exchange1

test.queue     

test.routingkey

 

由结果可以看出，由Exchange，Queue，RoutingKey三个才能决定一个从Exchange到Queue的唯一的线路。

 

 

技术图片

 

 

左边的Client向右边的Client发送消息，流程：

 

1，  获取Conection

 

2，  获取Channel

 

3，  定义Exchange，Queue

 

4，  使用一个RoutingKey将Queue Binding到一个Exchange上

 

5，  通过指定一个Exchange和一个RoutingKey来将消息发送到对应的Queue上，

 

6，  接收方在接收时也是获取connection，接着获取channel，然后指定一个Queue直接到它关心的Queue上取消息，它对Exchange，RoutingKey及如何binding都不关心，到对应的Queue上去取消息就OK了(这个是重点，接收方，只关心queue的名字即可，其他的都不关心！！！！！)

 

换句话说：

发送放关心：ExchangName， QueueName，BindingKey

接收方关心：QuueuName

 

一个Client发送消息，哪些Client可以收到消息，其核心就在于Exchange，RoutingKey，Queue的关系上。

 

 

Exchange

RoutingKey

Queue

1

E1

R1

Q1

2

 

R2

Q2

3

E2

R3

Q1

4

 

R4

Q2

5

E1

R5

Q1

6

E2

R6

Q1

 

我们可以这样理解，RoutingKey就像是个中间表，将两个表的数据进行多对多关联，只不过对于相同的Exchange和Queue，可以使用不同的RoutingKey重复关联多次。

以上是关于RabbitMQ基础整理的主要内容，如果未能解决你的问题，请参考以下文章

VS2015 代码片段整理

小程序各种功能代码片段整理---持续更新

常用python日期日志获取内容循环的代码片段

PHP项目开发经验整理

RabbitMQ详解------简介与安装

面试题整理