From：http://blog.csdn.net/column/details/rabbitmq.html

RabbitMQ 介绍

历史

RabbitMQ是一个由erlang开发的AMQP（Advanced Message Queue ）的开源实现。AMQP 的出现其实也是应了广大人民群众的需求，虽然在同步消息通讯的世界里有很多公开标准（如 COBAR的 IIOP ，或者是 SOAP 等），但是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的 MSMQ ，IBM 的 Websphere MQ 等），因此，在 2006 年的 6 月，Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。

RabbitMQ是由RabbitMQ Technologies Ltd开发并且提供商业支持的。该公司在2010年4月被SpringSource（VMWare的一个部门）收购。在2013年5月被并入Pivotal。其实VMWare，Pivotal和EMC本质上是一家的。不同的是VMWare是独立上市子公司，而Pivotal是整合了EMC的某些资源，现在并没有上市。

RabbitMQ的官网是：http://www.rabbitmq.com

应用场景

言归正传。RabbitMQ，或者说AMQP解决了什么问题，或者说它的应用场景是什么？

对于一个大型的软件系统来说，它会有很多的组件或者说模块或者说子系统或者（subsystem or Component or submodule）。那么这些模块的如何通信？这和传统的IPC有很大的区别。传统的IPC很多都是在单一系统上的，模块耦合性很大，不适合扩展（Scalability）；如果使用socket那么不同的模块的确可以部署到不同的机器上，但是还是有很多问题需要解决。比如：

1）信息的发送者和接收者如何维持这个连接，如果一方的连接中断，这期间的数据如何方式丢失？

2）如何降低发送者和接收者的耦合度？

3）如何让Priority高的接收者先接到数据？

4）如何做到load balance？有效均衡接收者的负载？

5）如何有效的将数据发送到相关的接收者？也就是说将接收者subscribe 不同的数据，如何做有效的filter。

6）如何做到可扩展，甚至将这个通信模块发到cluster上？

7）如何保证接收者接收到了完整，正确的数据？

AMDQ协议解决了以上的问题，而RabbitMQ实现了AMQP。

系统架构

成为系统架构可能不太合适，可能叫应用场景的系统架构更合适。

这个系统架构图版权属于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你的设计。

进一步的细节阐明

使用ack确认Message的正确传递

默认情况下，如果Message 已经被某个Consumer正确的接收到了，那么该Message就会被从queue中移除。当然也可以让同一个Message发送到很多的Consumer。

如果一个queue没被任何的Consumer Subscribe（订阅），那么，如果这个queue有数据到达，那么这个数据会被cache，不会被丢弃。当有Consumer时，这个数据会被立即发送到这个Consumer，这个数据被Consumer正确收到时，这个数据就被从queue中删除。

那么什么是正确收到呢？通过ack。每个Message都要被acknowledged（确认，ack）。我们可以显示的在程序中去ack，也可以自动的ack。如果有数据没有被ack，那么：

RabbitMQ Server会把这个信息发送到下一个Consumer。

如果这个app有bug，忘记了ack，那么RabbitMQ Server不会再发送数据给它，因为Server认为这个Consumer处理能力有限。

而且ack的机制可以起到限流的作用（Benefitto throttling）：在Consumer处理完成数据后发送ack，甚至在额外的延时后发送ack，将有效的balance Consumer的load。

当然对于实际的例子，比如我们可能会对某些数据进行merge，比如merge 4s内的数据，然后sleep 4s后再获取数据。特别是在监听系统的state，我们不希望所有的state实时的传递上去，而是希望有一定的延时。这样可以减少某些IO，而且终端用户也不会感觉到。

Reject a message

有两种方式，第一种的Reject可以让RabbitMQ Server将该Message 发送到下一个Consumer。第二种是从queue中立即删除该Message。

Creating a queue

Consumer和Procuder都可以通过 queue.declare 创建queue。对于某个Channel来说，Consumer不能declare一个queue，却订阅其他的queue。当然也可以创建私有的queue。这样只有app本身才可以使用这个queue。queue也可以自动删除，被标为auto-delete的queue在最后一个Consumer unsubscribe后就会被自动删除。那么如果是创建一个已经存在的queue呢？那么不会有任何的影响。需要注意的是没有任何的影响，也就是说第二次创建如果参数和第一次不一样，那么该操作虽然成功，但是queue的属性并不会被修改。

那么谁应该负责创建这个queue呢？是Consumer，还是Producer？

如果queue不存在，当然Consumer不会得到任何的Message。但是如果queue不存在，那么Producer Publish的Message会被丢弃。所以，还是为了数据不丢失，Consumer和Producer都try to create the queue！反正不管怎么样，这个接口都不会出问题。

queue对load balance的处理是完美的。对于多个Consumer来说，RabbitMQ 使用循环的方式（round-robin）的方式均衡的发送给不同的Consumer。

Exchanges

从架构图可以看出，Procuder Publish的Message进入了Exchange。接着通过“routing keys”， RabbitMQ会找到应该把这个Message放到哪个queue里。queue也是通过这个routing keys来做的绑定。

有三种类型的Exchanges：direct, fanout,topic。每个实现了不同的路由算法（routing algorithm）。

· Direct exchange: 如果 routing key 匹配, 那么Message就会被传递到相应的queue中。其实在queue创建时，它会自动的以queue的名字作为routing key来绑定那个exchange。

· Fanout exchange: 会向响应的queue广播。

· Topic exchange: 对key进行模式匹配，比如ab*可以传递到所有ab*的queue。

Virtual hosts

每个virtual host本质上都是一个RabbitMQ Server，拥有它自己的queue，exchagne，和bings rule等等。这保证了你可以在多个不同的application中使用RabbitMQ。

接下来我会使用Python来说明RabbitMQ的使用方法。

python 使用 RabbitMQ 写 "Hello World"

http://www.rabbitmq.com/tutorials/tutorial-one-python.html

使用Python（pika 0.9.8）实现从Producer到Consumer传递数据”Hello, World“。

RabbitMQ实现了AMQP定义的消息队列。它实现的功能”非常简单“：从Producer接收数据然后传递到Consumer。它能保证多并发，数据安全传递，可扩展。

和任何的Hello world一样，它们都不复杂。我们将会设计两个程序，一个发送Hello world，另一个接收这个数据并且打印到屏幕。
整体的设计如下图：

环境配置

RabbitMQ 实现了AMQP。因此，我们需要安装AMPQ的library。幸运的是对于多种编程语言都有实现。我们可以使用以下lib的任何一个：

py-amqplib
txAMQP
pika

在这里我们将使用pika. 可以通过 pip 包管理工具来安装:

$ sudo pip install pika==0.9.8

这个安装依赖于pip和Git-core。

On Ubuntu:

$ sudo apt-get install python-pip git-core

On Debian:

$ sudo apt-get install python-setuptools git-core
$ sudo easy_install pip

On Windows:To install easy_install, run the MS Windows Installer for setuptools
```
> easy_install pip
> pip install pika==0.9.8
```

Sending

第一个program send.py：发送Hello world 到queue。正如我们在上篇文章提到的，你程序的第一句话就是建立连接，第二句话就是创建channel：

#!/usr/bin/env python
import pika  connection = pika.BlockingConnection(pika.ConnectionParameters(  'localhost'))
channel = connection.channel()

创建连接传入的参数就是RabbitMQ Server的ip或者name。关于谁创建queue，上面也讨论过：Producer和Consumer都应该去创建。接下来我们创建名字为hello的queue：

channel.queue_declare(queue='hello')

创建了channel，我们可以通过相应的命令来list queue：

$ sudo rabbitmqctl list_queues
Listing queues ...
hello    0
...done.

现在我们已经准备好了发送了。

从架构图可以看出，Producer只能发送到exchange，它是不能直接发送到queue的。

现在我们使用默认的exchange（名字是空字符）。这个默认的exchange允许我们发送给指定的queue。routing_key就是指定的queue名字。

channel.basic_publish(exchange='',  routing_key='hello',  body='Hello World!')
print " [x] Sent 'Hello World!'"

退出前别忘了关闭connection。

connection.close()

Receiving

第二个program receive.py 将从queue中获取Message并且打印到屏幕。

第一步还是创建connection。第二步创建channel。第三步创建queue，name = hello：

channel.queue_declare(queue='hello')

接下来要subscribe了。在这之前，需要声明一个回调函数来处理接收到的数据。

def callback(ch, method, properties, body):  print " [x] Received %r" % (body,)

subscribe：

channel.basic_consume(callback,  queue='hello',  no_ack=True)

最后，准备好无限循环监听吧：

print ' [*] Waiting for messages. To exit press CTRL+C'
channel.start_consuming()

最终版本

send.py:

#!/usr/bin/env python
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()

receive.py:

#!/usr/bin/env python
import pika  connection = pika.BlockingConnection(pika.ConnectionParameters(  host='localhost'))
channel = connection.channel()  channel.queue_declare(queue='hello')  print ' [*] Waiting for messages. To exit press CTRL+C'  def callback(ch, method, properties, body):  print " [x] Received %r" % (body,)  channel.basic_consume(callback,  queue='hello',  no_ack=True)  channel.start_consuming()

最终运行

先运行 send.py program:

$ python send.py
[x] Sent 'Hello World!'

send.py 每次运行完都会停止。注意：现在数据已经存到queue里了。接收它：

$ python receive.py
[*] Waiting for messages. To exit press CTRL+C
[x] Received 'Hello World!'

任务分发机制

http://www.rabbitmq.com/tutorials/tutorial-two-Python.html

上面解决了从发送端（Producer）向接收端（Consumer）发送“Hello World”的问题。在实际的应用场景中，这是远远不够的。现在将结合更加实际的应用场景来讲解更多的高级用法。

当有Consumer需要大量的运算时，RabbitMQ Server需要一定的分发机制来balance每个Consumer的load。试想一下，对于web application来说，在一个很多的HTTP request里是没有时间来处理复杂的运算的，只能通过后台的一些工作线程来完成。接下来我们分布讲解。

应用场景就是RabbitMQ Server会将queue的Message分发给不同的Consumer以处理计算密集型的任务：

准备

在上面，我们简单在Message中包含了一个字符串"Hello World"。现在为了是Consumer做的是计算密集型的工作，那就不能简单的字符串了。在现实应用中，Consumer有可能做的是一个图片的resize，或者是pdf文件的渲染或者内容提取。但是作为Demo，还是用字符串模拟吧：通过字符串中的.的数量来决定计算的复杂度，每个.都会消耗1s，即sleep(1)。

还是复用上面的code，根据“计算密集型”做一下简单的修改，为了辨别，我们把send.py 的名字换成new_task.py

import sys  message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',  routing_key='hello',  body=message)
print " [x] Sent %r" % (message,)

同样的道理，把receive.py的名字换成worker.py，并且根据Message中的.的数量进行计算密集型模拟：

import time  def callback(ch, method, properties, body):  print " [x] Received %r" % (body,)  time.sleep( body.count('.') )  print " [x] Done"

Round-robin dispatching 循环分发

RabbitMQ的分发机制非常适合扩展，而且它是专门为并发程序设计的。如果现在load加重，那么只需要创建更多的Consumer来进行任务处理即可。当然了，对于负载还要加大怎么办？我没有遇到过这种情况，那就可以创建多个virtual Host，细化不同的通信类别了。

首先开启两个Consumer，即运行两个worker.py。

Console1：

shell1$ python worker.py  [*] Waiting for messages. To exit press CTRL+C

Consule2：

shell2$ python worker.py  [*] Waiting for messages. To exit press CTRL+C

Producer new_task.py要Publish Message了：

shell3$ python new_task.py First message.
shell3$ python new_task.py Second message..
shell3$ python new_task.py Third message...
shell3$ python new_task.py Fourth message....
shell3$ python new_task.py Fifth message.....

注意一下：.代表的sleep(1)。接着开一下Consumer worker.py收到了什么:

Console1：

shell1$ python worker.py  [*] Waiting for messages. To exit press CTRL+C  [x] Received 'First message.'  [x] Received 'Third message...'  [x] Received 'Fifth message.....'

Console2：

shell2$ python worker.py  [*] Waiting for messages. To exit press CTRL+C  [x] Received 'Second message..'  [x] Received 'Fourth message....'

默认情况下，RabbitMQ 会顺序的分发每个Message。当每个收到ack后，会将该Message删除，然后将下一个Message分发到下一个Consumer。这种分发方式叫做round-robin。这种分发还有问题，接着向下读吧。

Message acknowledgment 消息确认

每个Consumer可能需要一段时间才能处理完收到的数据。如果在这个过程中，Consumer出错了，异常退出了，而数据还没有处理完成，那么非常不幸，这段数据就丢失了。因为我们采用no-ack的方式进行确认，也就是说，每次Consumer接到数据后，而不管是否处理完成，RabbitMQ Server会立即把这个Message标记为完成，然后从queue中删除了。

如果一个Consumer异常退出了，它处理的数据能够被另外的Consumer处理，这样数据在这种情况下就不会丢失了（注意是这种情况下）。

为了保证数据不被丢失，RabbitMQ支持消息确认机制，即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到，那么我们不能采用no-ack。而应该是在处理完数据后发送ack。

在处理数据后发送的ack，就是告诉RabbitMQ数据已经被接收，处理完成，RabbitMQ可以去安全的删除它了。

如果Consumer退出了但是没有发送ack，那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。

这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说，RabbitMQ给了Consumer足够长的时间来做数据处理。

默认情况下，消息确认是打开的（enabled）。在上篇文章中我们通过no_ack = True 关闭了ack。重新修改一下callback，以在消息处理完成后发送ack：

def callback(ch, method, properties, body):  print " [x] Received %r" % (body,)  time.sleep( body.count('.') )  print " [x] Done"  ch.basic_ack(delivery_tag = method.delivery_tag)  channel.basic_consume(callback,  queue='hello')

这样即使你通过Ctr-C中断了worker.py，那么Message也不会丢失了，它会被分发到下一个Consumer。

如果忘记了ack，那么后果很严重。当Consumer退出时，Message会重新分发。然后RabbitMQ会占用越来越多的内存，由于RabbitMQ会长时间运行，因此这个“内存泄漏”是致命的。去调试这种错误，可以通过一下命令打印un-acked Messages：

$ sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged
Listing queues ...
hello    0       0
...done.

Message durability消息持久化

在上一节中我们知道了即使Consumer异常退出，Message也不会丢失。但是如果RabbitMQ Server退出呢？软件都有bug，即使RabbitMQ Server是完美毫无bug的（当然这是不可能的，是软件就有bug，没有bug的那不叫软件），它还是有可能退出的：被其它软件影响，或者系统重启了，系统panic了。。。

为了保证在RabbitMQ退出或者crash了数据仍没有丢失，需要将queue和Message都要持久化。

queue的持久化需要在声明时指定durable=True：

channel.queue_declare(queue='hello', durable=True)

上述语句执行不会有什么错误，但是确得不到我们想要的结果，原因就是RabbitMQ Server已经维护了一个叫hello的queue，那么上述执行不会有任何的作用，也就是hello的任何属性都不会被影响。这一点在上篇文章也讨论过。

那么workaround也很简单，声明一个另外的名字的queue，比如名字定位task_queue：

[python] view plain copy

channel.queue_declare(queue='task_queue', durable=True)

再次强调，Producer和Consumer都应该去创建这个queue，尽管只有一个地方的创建是真正起作用的：

接下来，需要持久化Message，即在Publish的时候指定一个properties，方式如下：

[python] view plain copy

channel.basic_publish(exchange='',
routing_key="task_queue",
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))

关于持久化的进一步讨论：

为了数据不丢失，我们采用了：

在数据处理结束后发送ack，这样RabbitMQ Server会认为Message Deliver 成功。
持久化queue，可以防止RabbitMQ Server 重启或者crash引起的数据丢失。
持久化Message，理由同上。

但是这样能保证数据100%不丢失吗？

答案是否定的。问题就在与RabbitMQ需要时间去把这些信息存到磁盘上，这个time window虽然短，但是它的确还是有。在这个时间窗口内如果数据没有保存，数据还会丢失。还有另一个原因就是RabbitMQ并不是为每个Message都做fsync：它可能仅仅是把它保存到Cache里，还没来得及保存到物理磁盘上。

因此这个持久化还是有问题。但是对于大多数应用来说，这已经足够了。当然为了保持一致性，你可以把每次的publish放到一个transaction中。这个transaction的实现需要user defined codes。

那么商业系统会做什么呢？一种可能的方案是在系统panic时或者异常重启时或者断电时，应该给各个应用留出时间去flash cache，保证每个应用都能exit gracefully。

Fair dispatch 公平分发

你可能也注意到了，分发机制不是那么优雅。默认状态下，RabbitMQ将第n个Message分发给第n个Consumer。当然n是取余后的。它不管Consumer是否还有unacked Message，只是按照这个默认机制进行分发。

那么如果有个Consumer工作比较重，那么就会导致有的Consumer基本没事可做，有的Consumer却是毫无休息的机会。那么，RabbitMQ是如何处理这种问题呢？

通过 basic.qos 方法设置prefetch_count=1 。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。设置方法如下：

[python] view plain copy

channel.basic_qos(prefetch_count=1)

注意，这种方法可能会导致queue满。当然，这种情况下你可能需要添加更多的Consumer，或者创建更多的virtualHost来细化你的设计。

最终版本

new_task.py script:

[python] view plain copy

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))
print " [x] Sent %r" % (message,)
connection.close()

worker.py script:

[python] view plain copy

#!/usr/bin/env python
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
print ' [*] Waiting for messages. To exit press CTRL+C'
def callback(ch, method, properties, body):
print " [x] Received %r" % (body,)
time.sleep( body.count('.') )
print " [x] Done"
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
queue='task_queue')
channel.start_consuming()

分发到多Consumer（Publish/Subscribe）

http://www.rabbitmq.com/tutorials/tutorial-three-Python.html

上篇文章中，我们把每个Message都是deliver到某个Consumer。在这篇文章中，我们将会将同一个Message deliver到多个Consumer中。这个模式也被成为 "publish / subscribe"。
这篇文章中，我们将创建一个日志系统，它包含两个部分：第一个部分是发出log（Producer），第二个部分接收到并打印（Consumer）。我们将构建两个Consumer，第一个将log写到物理磁盘上；第二个将log输出的屏幕。

Exchanges

RabbitMQ 的Messaging Model就是Producer并不会直接发送Message到queue。实际上，Producer并不知道它发送的Message是否已经到达queue。

Producer发送的Message实际上是发到了Exchange中。它的功能也很简单：从Producer接收Message，然后投递到queue中。Exchange需要知道如何处理Message，是把它放到那个queue中，还是放到多个queue中？这个rule是通过Exchange 的类型定义的。

我们知道有三种类型的Exchange：direct, topic 和fanout。fanout就是广播模式，会将所有的Message都放到它所知道的queue中。创建一个名字为logs，类型为fanout的Exchange：

[python] view plain copy

channel.exchange_declare(exchange='logs',
type='fanout')

Listing exchanges

通过rabbitmqctl可以列出当前所有的Exchange：

[python] view plain copy

$ sudo rabbitmqctl list_exchanges
Listing exchanges ...
logs fanout
amq.direct direct
amq.topic topic
amq.fanout fanout
amq.headers headers
...done.

注意 amq.* exchanges 和the default (unnamed)exchange是RabbitMQ默认创建的。

现在我们可以通过exchange，而不是routing_key来publish Message了：

[python] view plain copy

channel.basic_publish(exchange='logs',
routing_key='',
body=message)

Temporary queues

截至现在，我们用的queue都是有名字的：第一个是hello，第二个是task_queue。使用有名字的queue，使得在Producer和Consumer之前共享queue成为可能。

但是对于我们将要构建的日志系统，并不需要有名字的queue。我们希望得到所有的log，而不是它们中间的一部分。而且我们只对当前的log感兴趣。为了实现这个目标，我们需要两件事情：
1）每当Consumer连接时，我们需要一个新的，空的queue。因为我们不对老的log感兴趣。幸运的是，如果在声明queue时不指定名字，那么RabbitMQ会随机为我们选择这个名字。方法：

[python] view plain copy

result = channel.queue_declare()

通过result.method.queue 可以取得queue的名字。基本上都是这个样子：amq.gen-JzTY20BRgKO-HjmUJj0wLg。
2）当Consumer关闭连接时，这个queue要被deleted。可以加个exclusive的参数。方法：

[python] view plain copy

result = channel.queue_declare(exclusive=True)

Bindings绑定

现在我们已经创建了fanout类型的exchange和没有名字的queue（实际上是RabbitMQ帮我们取了名字）。那exchange怎么样知道它的Message发送到哪个queue呢？答案就是通过bindings：绑定。

方法：

[python] view plain copy

channel.queue_bind(exchange='logs',
queue=result.method.queue)

现在logs的exchange就将它的Message附加到我们创建的queue了。

Listing bindings

使用命令rabbitmqctl list_bindings。

最终版本

我们最终实现的数据流图如下：

Producer，在这里就是产生log的program，基本上和前几个都差不多。最主要的区别就是publish通过了exchange而不是routing_key。

emit_log.py script:

[python] view plain copy

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
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 " [x] Sent %r" % (message,)
connection.close()

还有一点要注意的是我们声明了exchange。publish到一个不存在的exchange是被禁止的。如果没有queue bindings exchange的话，log是被丢弃的。
Consumer：receive_logs.py:

[python] view plain copy

#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
type='fanout')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange='logs',
queue=queue_name)
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()

我们开始不是说需要两个Consumer吗？一个负责记录到文件；一个负责打印到屏幕？
其实用重定向就可以了，当然你想修改callback自己写文件也行。我们使用重定向的方法：
We're done. If you want to save logs to a file, just open a console and type:

[python] view plain copy

$ python receive_logs.py > logs_from_rabbit.log

Consumer2：打印到屏幕：

[python] view plain copy

$ python receive_logs.py

接下来，Producer：

[python] view plain copy

$ python emit_log.py

使用命令rabbitmqctl list_bindings你可以看我们创建的queue。
一个output：

[python] view plain copy

$ sudo rabbitmqctl list_bindings
Listing bindings ...
logs exchange amq.gen-JzTY20BRgKO-HjmUJj0wLg queue []
logs exchange amq.gen-vso0PVvyiRIL2WoV3i48Yg queue []
...done.

这个结果还是很好理解的。

Routing 消息路由

http://www.rabbitmq.com/tutorials/tutorial-four-Python.html

上篇文章中，我们构建了一个简单的日志系统。接下来，我们将丰富它：能够使用不同的severity来监听不同等级的log。比如我们希望只有error的log才保存到磁盘上。

Bindings绑定

上篇文章中我们是这么做的绑定：

[python] view plain copy

channel.queue_bind(exchange=exchange_name,
queue=queue_name)

绑定其实就是关联了exchange和queue。或者这么说：queue对exchagne的内容感兴趣，exchange要把它的Message deliver到queue中。

实际上，绑定可以带routing_key 这个参数。其实这个参数的名称和basic_publish 的参数名是相同了。为了避免混淆，我们把它成为binding key。
使用一个key来创建binding :

[python] view plain copy

channel.queue_bind(exchange=exchange_name,
queue=queue_name,
routing_key='black')

对于fanout的exchange来说，这个参数是被忽略的。

Direct exchange

Direct exchange的路由算法非常简单：通过binding key的完全匹配，可以通过下图来说明。

exchange X和两个queue绑定在一起。Q1的binding key是orange。Q2的binding key是black和green。
当P publish key是orange时，exchange会把它放到Q1。如果是black或者green那么就会到Q2。其余的Message都会被丢弃。

Multiple bindings

多个queue绑定同一个key是可以的。对于下图的例子，Q1和Q2都绑定了black。也就是说，对于routing key是black的Message，会被deliver到Q1和Q2。其余的Message都会被丢弃。

Emitting logs

首先是我们要创建一个direct的exchange：

[python] view plain copy

channel.exchange_declare(exchange='direct_logs',
type='direct')

我们将使用log的severity作为routing key，这样Consumer可以针对不同severity的log进行不同的处理。
publish：

[python] view plain copy

channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)

我们使用三种severity：'info', 'warning', 'error'.

Subscribing

对于queue，我们需要绑定severity：

[python] view plain copy

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
for severity in severities:
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)

最终版本

The code for emit_log_direct.py:

[python] view plain copy

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
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 " [x] Sent %r:%r" % (severity, message)
connection.close()

The code for receive_logs_direct.py:

[python] view plain copy

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
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:
print >> sys.stderr, "Usage: %s [info] [warning] [error]" % \
(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()

我们想把warning和error的log记录到一个文件中：

[python] view plain copy

$ python receive_logs_direct.py warning error > logs_from_rabbit.log

打印所有log到屏幕：

[python] view plain copy

$ python receive_logs_direct.py info warning error
[*] Waiting for logs. To exit press CTRL+C

使用主题进行消息分发

http://www.rabbitmq.com/tutorials/tutorial-five-Python.html

上面实现了一个简单的日志系统。Consumer可以监听不同severity的log。但是，这也是它之所以叫做简单日志系统的原因，因为是仅仅能够通过severity设定。不支持更多的标准。

比如syslog unix的日志工具，它可以通过severity (info/warn/crit...) 和模块(auth/cron/kern...)。这可能更是我们想要的：我们可以仅仅需要cron模块的log。

为了实现类似的功能，我们需要用到topic exchange。

Topic exchange

对于Message的routing_key是有限制的，不能使任意的。格式是以点号“."分割的字符表。比如："stock.usd.nyse", "nyse.vmw", "quick.orange.rabbit"。你可以放任意的key在routing_key中，当然最长不能超过255 bytes。

对于routing_key，有两个特殊字符（在正则表达式里叫元字符）：

* (星号) 代表任意一个单词
# (hash) 0个或者多个单词

请看下面一个例子:

Producer发送消息时需要设置routing_key，routing_key包含三个单词和两个点号。第一个key是描述了celerity（灵巧，敏捷），第二个是colour（色彩），第三个是species（物种）："<celerity>.<colour>.<species>"。

在这里我们创建了两个绑定： Q1 的binding key 是"*.orange.*"； Q2 是 "*.*.rabbit" 和 "lazy.#"：

Q1 感兴趣所有orange颜色的动物
Q2 感兴趣所有的rabbits和所有的lazy的

比如routing_key是 "quick.orange.rabbit"将会发送到Q1和Q2中。消息"lazy.orange.elephant" 也会发送到Q1和Q2。但是"quick.orange.fox" 会发送到Q1；"lazy.brown.fox"会发送到Q2。"lazy.pink.rabbit" 也会发送到Q2，但是尽管两个routing_key都匹配，它也只是发送一次。"quick.brown.fox" 会被丢弃。

如果发送的单词不是3个呢？答案要看情况，因为#是可以匹配0个或任意个单词。比如"orange" or "quick.orange.male.rabbit"，它们会被丢弃。如果是lazy那么就会进入Q2。类似的还有 "lazy.orange.male.rabbit"，尽管它包含四个单词。

Topic exchange和其他exchange

由于有"*" (star) and "#" (hash)， Topic exchange 非常强大并且可以转化为其他的exchange：

如果binding_key 是 "#" - 它会接收所有的Message，不管routing_key是什么，就像是fanout exchange。

如果 "*" (star) and "#" (hash) 没有被使用，那么topic exchange就变成了direct exchange。

代码实现

现在我们要refine我们上篇的日志系统。routing keys 有两个部分: "<facility>.<severity>"。

The code for emit_log_topic.py:

[python] view plain copy

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
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)
connection.close()

The code for receive_logs_topic.py:

[python] view plain copy

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
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.stderr, "Usage: %s [binding_key]..." % (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()

运行和结果

接收所有的log:

[python] view plain copy

python receive_logs_topic.py "#"

接收所有kern facility的log：

[python] view plain copy

python receive_logs_topic.py "kern.*"

仅仅接收critical的log：

[python] view plain copy

python receive_logs_topic.py "*.critical"

可以创建多个绑定：

[python] view plain copy

python receive_logs_topic.py "kern.*" "*.critical"

Producer产生一个log："kern.critical" type:

[python] view plain copy

python emit_log_topic.py "kern.critical" "A critical kernel error"

课后思考题:

Will "*" binding catch a message sent with an empty routing key?
Will "#.*" catch a message with a string ".." as a key? Will it catch a message with a single word key?
How different is "a.*.#" from "a.#"?

适用于云计算集群的远程调用(RPC)

http://www.rabbitmq.com/tutorials/tutorial-six-Python.html

在云计算环境中，很多时候需要用它其他机器的计算资源，我们有可能会在接收到Message进行处理时，会把一部分计算任务分配到其他节点来完成。那么，RabbitMQ如何使用RPC呢？在本篇文章中，我们将会通过其它节点求来斐波纳契完成示例。

客户端接口 Client interface

为了展示一个RPC服务是如何使用的，我们将创建一段很简单的客户端class。它将会向外提供名字为call的函数，这个call会发送RPC请求并且阻塞知道收到RPC运算的结果。代码如下：

[python] view plain copy

fibonacci_rpc = FibonacciRpcClient()
result = fibonacci_rpc.call(4)
print "fib(4) is %r" % (result,)

回调函数队列 Callback queue

总体来说，在RabbitMQ进行RPC远程调用是比较容易的。client发送请求的Message然后server返回响应结果。为了收到响应client在publish message时需要提供一个”callback“（回调）的queue地址。code如下:

[python] view plain copy

result = channel.queue_declare(exclusive=True)
callback_queue = result.method.queue
channel.basic_publish(exchange='',
routing_key='rpc_queue',
properties=pika.BasicProperties(
reply_to = callback_queue,
),
body=request)
# ... and some code to read a response message from the callback_queue ...

Message properties

AMQP 预定义了14个属性。它们中的绝大多很少会用到。以下几个是平时用的比较多的：

delivery_mode: 持久化一个Message（通过设定值为2）。其他任意值都是非持久化。请移步RabbitMQ消息队列（三）：任务分发机制
content_type: 描述mime-type 的encoding。比如设置为JSON编码：设置该property为application/json。
reply_to: 一般用来指明用于回调的queue（Commonly used to name a callback queue）。
correlation_id: 在请求中关联处理RPC响应（correlate RPC responses with requests）。

总结

工作流程：

当客户端启动时，它创建了匿名的exclusive callback queue.
客户端的RPC请求时将同时设置两个properties： reply_to设置为callback queue；correlation_id设置为每个request一个独一无二的值.
请求将被发送到an rpc_queue queue.
RPC端或者说server一直在等待那个queue的请求。当请求到达时，它将通过在reply_to指定的queue回复一个message给client。
client一直等待callback queue的数据。当message到达时，它将检查correlation_id的值，如果值和它request发送时的一致那么就将返回响应。

最终实现

The code for rpc_server.py:

[python] view plain copy

#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='rpc_queue')
def fib(n):
if n == 0:
return 0
elif n == 1:
return 1
else:
return fib(n-1) + fib(n-2)
def on_request(ch, method, props, body):
n = int(body)
print " [.] fib(%s)" % (n,)
response = fib(n)
ch.basic_publish(exchange='',
routing_key=props.reply_to,
properties=pika.BasicProperties(correlation_id = \
props.correlation_id),
body=str(response))
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue='rpc_queue')
print " [x] Awaiting RPC requests"
channel.start_consuming()

The server code is rather straightforward:

(4) As usual we start by establishing the connection and declaring the queue.
(11) We declare our fibonacci function. It assumes only valid positive integer input. (Don't expect this one to work for big numbers, it's probably the slowest recursive implementation possible).
(19) We declare a callback for basic_consume, the core of the RPC server. It's executed when the request is received. It does the work and sends the response back.
(32) We might want to run more than one server process. In order to spread the load equally over multiple servers we need to set theprefetch_count setting.

The code for rpc_client.py:

[python] view plain copy

#!/usr/bin/env python
import pika
import uuid
class FibonacciRpcClient(object):
def __init__(self):
self.connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
self.channel = self.connection.channel()
result = self.channel.queue_declare(exclusive=True)
self.callback_queue = result.method.queue
self.channel.basic_consume(self.on_response, no_ack=True,
queue=self.callback_queue)
def on_response(self, ch, method, props, body):
if self.corr_id == props.correlation_id:
self.response = body
def call(self, n):
self.response = None
self.corr_id = str(uuid.uuid4())
self.channel.basic_publish(exchange='',
routing_key='rpc_queue',
properties=pika.BasicProperties(
reply_to = self.callback_queue,
correlation_id = self.corr_id,
),
body=str(n))
while self.response is None:
self.connection.process_data_events()
return int(self.response)
fibonacci_rpc = FibonacciRpcClient()
print " [x] Requesting fib(30)"
response = fibonacci_rpc.call(30)
print " [.] Got %r" % (response,)

The client code is slightly more involved:

(7) We establish a connection, channel and declare an exclusive 'callback' queue for replies.
(16) We subscribe to the 'callback' queue, so that we can receive RPC responses.
(18) The 'on_response' callback executed on every response is doing a very simple job, for every response message it checks if thecorrelation_id is the one we're looking for. If so, it saves the response inself.response and breaks the consuming loop.
(23) Next, we define our main call method - it does the actual RPC request.
(24) In this method, first we generate a unique correlation_id number and save it - the 'on_response' callback function will use this value to catch the appropriate response.
(25) Next, we publish the request message, with two properties: reply_to and correlation_id.
(32) At this point we can sit back and wait until the proper response arrives.
(33) And finally we return the response back to the user.

开始rpc_server.py：

[python] view plain copy

$ python rpc_server.py
[x] Awaiting RPC requests

通过client来请求fibonacci数:

[python] view plain copy

$ python rpc_client.py
[x] Requesting fib(30)

现在这个设计并不是唯一的，但是这个实现有以下优势：

如何RPC server太慢，你可以扩展它:启动另外一个RPC server。
在client端，无所进行加锁能同步操作，他所作的就是发送请求等待响应。

我们的code还是挺简单的，并没有尝试去解决更复杂和重要的问题，比如：

如果没有server在运行，client需要怎么做？
RPC应该设置超时机制吗?
如果server运行出错并且抛出了异常，需要将这个问题转发到client吗？
需要边界检查吗？

RabbitMQ消息队列的小伙伴: ProtoBuf（Google Protocol Buffer）

什么是ProtoBuf？

一种轻便高效的结构化数据存储格式，可以用于结构化数据串行化，或者说序列化。它很适合做数据存储或 RPC 数据交换格式。可用于通讯协议、数据存储等领域的语言无关、平台无关、可扩展的序列化结构数据格式。目前提供了 C++、Java、Python 三种语言的 API。

它可以作为RabbitMQ的Message的数据格式进行传输，由于是结构化的数据，这样就极大的方便了Consumer的数据高效处理。当然了你可能说使用XML不也可以吗？与XML相比，ProtoBuf有以下优势：

简单
size小了3-10倍
速度快乐20-100倍
易于编程
减小了语义的歧义

当然了，的确还有很多类似的技术，比如JSON，Thrift等等，和他们相比，ProtoBuf的优势或者劣势在哪里？简单说来，ProtoBuf就是简单，快。以测试为证：项目 thrift-protobuf-compare 比较了这些类似的技术，下图显示了该项目的一项测试结果。

在占用空间上的性能比较：

由此可见，ProtoBuf具有速度和空间的优势，使得它现在应用非常广泛。比如Hadoop就使用了它。

更多信息，请阅 http://www.ibm.com/developerworks/cn/Linux/l-cn-gpb/。

Publisher的消息确认机制

在前面的文章中提到了queue和consumer之间的消息确认机制：通过设置ack。那么Publisher能不到知道他post的Message有没有到达queue，甚至更近一步，是否被某个Consumer处理呢？毕竟对于一些非常重要的数据，可能Publisher需要确认某个消息已经被正确处理。

在我们的系统中，我们没有是实现这种确认，也就是说，不管Message是否被Consume了，Publisher不会去care。他只是将自己的状态publish给上层，由上层的逻辑去处理。如果Message没有被正确处理，可能会导致某些状态丢失。但是由于提供了其他强制刷新全部状态的机制，因此这种异常情况的影响也就可以忽略不计了。

对于某些异步操作，比如客户端需要创建一个FileSystem，这个可能需要比较长的时间，甚至要数秒钟。这时候通过RPC可以解决这个问题。因此也就不存在Publisher端的确认机制了。

那么，有没有一种机制能保证Publisher能够感知它的Message有没有被处理的？答案肯定的。在这里感谢笑天居士同学：他在我的《RabbitMQ消息队列（三）：任务分发机制》文后留言一起讨论了问题，而且也查找了一些资料。在这里我整理了一下他转载和一篇文章和原创的一篇文章。衔接已经附后。

事务机制 VS Publisher Confirm

如果采用标准的 AMQP 协议，则唯一能够保证消息不会丢失的方式是利用事务机制 -- 令 channel 处于 transactional 模式、向其 publish 消息、执行 commit 动作。在这种方式下，事务机制会带来大量的多余开销，并会导致吞吐量下降 250% 。为了补救事务带来的问题，引入了 confirmation 机制（即 Publisher Confirm）。

为了使能 confirm 机制，client 首先要发送 confirm.select 方法帧。取决于是否设置了 no-wait 属性，broker 会相应的判定是否以 confirm.select-ok 进行应答。一旦在 channel 上使用 confirm.select方法，channel 就将处于 confirm 模式。处于 transactional 模式的 channel 不能再被设置成 confirm 模式，反之亦然。
一旦 channel 处于 confirm 模式，broker 和 client 都将启动消息计数（以 confirm.select 为基础从 1 开始计数）。broker 会在处理完消息后，在当前 channel 上通过发送 basic.ack 的方式对其进行 confirm 。delivery-tag 域的值标识了被 confirm 消息的序列号。broker 也可以通过设置 basic.ack 中的 multiple 域来表明到指定序列号为止的所有消息都已被 broker 正确的处理了。

在异常情况中，broker 将无法成功处理相应的消息，此时 broker 将发送 basic.nack 来代替 basic.ack 。在这个情形下，basic.nack 中各域值的含义与 basic.ack 中相应各域含义是相同的，同时 requeue 域的值应该被忽略。通过 nack 一或多条消息，broker 表明自身无法对相应消息完成处理，并拒绝为这些消息的处理负责。在这种情况下，client 可以选择将消息 re-publish 。

在 channel 被设置成 confirm 模式之后，所有被 publish 的后续消息都将被 confirm（即 ack）或者被 nack 一次。但是没有对消息被 confirm 的快慢做任何保证，并且同一条消息不会既被 confirm 又被 nack 。

消息在什么时候确认

broker 将在下面的情况中对消息进行 confirm ：

broker 发现当前消息无法被路由到指定的 queues 中（如果设置了 mandatory 属性，则 broker 会先发送 basic.return）
非持久属性的消息到达了其所应该到达的所有 queue 中（和镜像 queue 中）
持久消息到达了其所应该到达的所有 queue 中（和镜像 queue 中），并被持久化到了磁盘（被 fsync）
持久消息从其所在的所有 queue 中被 consume 了（如果必要则会被 acknowledge）

broker 会丢失持久化消息，如果 broker 在将上述消息写入磁盘前异常。在一定条件下，这种情况会导致 broker 以一种奇怪的方式运行。例如，考虑下述情景：

1. 一个 client 将持久消息 publish 到持久 queue 中
   2. 另一个 client 从 queue 中 consume 消息（注意：该消息具有持久属性，并且 queue 是持久化的），当尚未对其进行 ack
   3. broker 异常重启
   4. client 重连并开始 consume 消息

在上述情景下，client 有理由认为消息需要被（broker）重新 deliver 。但这并非事实：重启（有可能）会令 broker 丢失消息。为了确保持久性，client 应该使用 confirm 机制。如果 publisher 使用的 channel 被设置为 confirm 模式，publisher 将不会收到已丢失消息的 ack（这是因为 consumer 没有对消息进行 ack ，同时该消息也未被写入磁盘）。

编程实现

首先要区别AMQP协议mandatory和immediate标志位的作用。

mandatory和immediate是AMQP协议中basic.pulish方法中的两个标志位，它们都有当消息传递过程中不可达目的地时将消息返回给生产者的功能。具体区别在于：
1. mandatory标志位
当mandatory标志位设置为true时，如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue，那么会调用basic.return方法将消息返还给生产者；当mandatory设为false时，出现上述情形broker会直接将消息扔掉。
2. immediate标志位
当immediate标志位设置为true时，如果exchange在将消息route到queue(s)时发现对应的queue上没有消费者，那么这条消息不会放入队列中。当与消息routeKey关联的所有queue(一个或多个)都没有消费者时，该消息会通过basic.return方法返还给生产者。

具体的代码参考请参考参考资料1.

参考资料：

1. http://blog.csdn.NET/jiao_fuyou/article/details/21594205

2. http://blog.csdn.net/jiao_fuyou/article/details/21594947

3. http://my.oschina.Net/moooofly/blog/142095

RabbitMQ从入门到精通相关推荐

RabbitMQ 从入门到精通（一）
目录 1. 初识RabbitMQ 2. AMQP 3.RabbitMQ的极速入门 4. Exchange(交换机)详解 4.1 Direct Exchange 4.2 Topic Exchange 4 ...
RabbitMQ 从入门到精通消息应答持久化交换机队列发布确认集群等
RabbitMQ消息队列 RabbitMQ 的概念 RabbitMQ 是一个消息中间件:它接受并转发消息.你可以把它当做一个快递站点,当你要发送一个包裹时,你把你的包裹放到快递站,快递员最终会把你的快 ...
SpringBoot从入门到精通教程
SpringBoot从入门到精通教程一.来自ImportNew公众号的SpringBoot教程系列,可参考学习 SpringBoot (一) :入门篇--http://mp.weixin.qq.co ...
Iaas-openstack从入门到精通
学习课程连接: openstack从入门到精通第二章 openstack核心组件详细介绍 2-1openstack和kubernetes联系和区别 2-2准备安装openstack需要实验环境虚拟 ...
ActiveMQ从入门到精通（全）
目录前言 1. 定义 2. 安装及配置 2.1 控制台访问 2.2 新建项目 2.3 配置文件 3. 队列案例 3.1 JMS编码 3.2 消息生产者 3.3 消息消费者 3.4 监听式消费者 3. ...
SpringCloud Alibaba 从入门到精通（精选）
SpringCloud Alibaba 从入门到精通一. 课程介绍 1.1 课程导学 1.2 项目环境搭建二. SpringBoot基础 2.1 本章概述 2.2 Spring Boot是什么?能 ...
超硬核Java工程师学习指南，真正的从入门到精通，众多粉丝亲测已拿offer！
最近有很多小伙伴来问我,Java小白如何入门,如何安排学习路线,每一步应该怎么走比较好.原本我以为之前的几篇文章已经可以解决大家的问题了,其实不然,因为我之前写的文章都是站在Java后端的全局上进行思 ...
java架构师入门教程,java技术架构师入门到精通高薪就业教程百度云下载
java技术架构师入门到精通高薪就业视频教程百度云课程目录: JAVA架构课开班典礼 JVM性能调优专题 JVM整体结构深度解析 JVM内存分配机制详解(此视频作废) JVM字节码文件结构深度剖析 ...
Elasticsearch7从入门到精通(简介、部署、原理、开发、ELK)
Elasticsearch7从入门到精通(简介.部署.原理.开发.ELK) 第1章.Elasticsearch简介 1-1.Elasticsearch介绍 Elasticsearch官方网站:http ...

RabbitMQ从入门到精通

RabbitMQ 介绍

历史

应用场景

系统架构

进一步的细节阐明

使用ack确认Message的正确传递

Reject a message

Creating a queue

Exchanges

Virtual hosts

python 使用 RabbitMQ 写 "Hello World"

环境配置

Sending

Receiving

最终版本

最终运行

任务分发机制

准备

Round-robin dispatching 循环分发

Message acknowledgment 消息确认

Message durability消息持久化

关于持久化的进一步讨论：

Fair dispatch 公平分发

最终版本

分发到多Consumer（Publish/Subscribe）

Exchanges

Temporary queues

Bindings绑定

最终版本

Routing 消息路由

Bindings绑定

Direct exchange

Multiple bindings

Emitting logs

Subscribing

最终版本

使用主题进行消息分发

Topic exchange

代码实现

运行和结果

适用于云计算集群的远程调用(RPC)

客户端接口 Client interface

回调函数队列 Callback queue

Message properties

相关id Correlation id

总结

最终实现

RabbitMQ消息队列的小伙伴: ProtoBuf（Google Protocol Buffer）

Publisher的消息确认机制

事务机制 VS Publisher Confirm

消息在什么时候确认

编程实现

RabbitMQ从入门到精通相关推荐

最新文章

热门文章