RabbitMQ|RabbitMQ 四种类型发送接收数据方式 RabbitMQ|RabbitMQ类型

1.基本用法
生产者

文章图片

文章图片

消费者

文章图片

文章图片
2.实现功能：（1）rabbitmq循环调度，将消息循环发送给不同的消费者，如：消息1,3,5发送给消费者1；消息2,4,6发送给消费者2。
（2）消息确认机制，为了确保一个消息不会丢失，RabbitMQ支持消息的确认 , 一个 ack(acknowlegement) 是从消费者端发送一个确认去告诉RabbitMQ 消息已经接收了、处理了，RabbitMQ可以释放并删除掉了。如果一个消费者死掉了（channel关闭、connection关闭、或者TCP连接断开了）而没有发送ack，RabbitMQ 就会认为这个消息没有被消费者处理，并会重新发送到生产者的队列里，如果同时有另外一个消费者在线，rabbitmq将会将消息很快转发到另外一个消费者中。那样的话你就能确保虽然一个消费者死掉，但消息不会丢失。
这个是没有超时的，当消费方（consumer）死掉后RabbitMQ会重新转发消息，即使处理这个消息需要很长很长时间也没有问题。消息的 acknowlegments 默认是打开的，在前面的例子中关闭了： no_ack = True . 现在删除这个标识然后发送一个 acknowledgment。
（3）消息持久化，将消息写入硬盘中。RabbitMQ不允许你重新定义一个已经存在、但属性不同的queue。需要标记消息为持久化的 - 要通过设置 delivery_mode 属性为 2来实现。
消息持久化的注意点：
标记消息为持久化并不能完全保证消息不会丢失，尽管已经告诉RabbitMQ将消息保存到磁盘，但RabbitMQ接收到的消息在还没有保存的时候，仍然有一个短暂的时间窗口。RabbitMQ不会对每个消息都执行同步 --- 可能只是保存到缓存cache还没有写入到磁盘中。因此这个持久化保证并不是很强，但这比我们简单的任务queue要好很多，如果想要很强的持久化保证，可以使用 publisher confirms。
（4）公平调度。在一个消费者未处理完一个消息之前不要分发新的消息给它，而是将这个新消息分发给另一个不是很忙的消费者进行处理。为了解决这个问题我们可以在消费者代码中使用 channel.basic.qos ( prefetch_count = 1 )，将消费者设置为公平调度。
生产者

文章图片

文章图片
消费者

文章图片

文章图片
3.交换机
exchange：交换机。生产者不是将消息发送给队列，而是将消息发送给交换机，由交换机决定将消息发送给哪个队列。所以exchange必须准确知道消息是要送到哪个队列，还是要被丢弃。因此要在exchange中给exchange定义规则，所有的规则都是在exchange的类型中定义的。
exchange有4个类型：direct, topic, headers ,fanout
之前，我们并没有讲过exchange，但是我们仍然可以将消息发送到队列中。这是因为我们用的是默认exchange.也就是说之前写的：exchange=''，空字符串表示默认的exchange。
之前的代码结构：

exchange = '参数'
参数表示exchange 的名字，空字符串是默认或者没有exchange。消息被路由到某队列的根据是：routing_key.。如果routing_key的值存在的话。
现在，我们可以用我们自己命名的exchange来代替默认的exchange。

（1）fanout：广播类型，生产者将消息发送给所有消费者，如果某个消费者没有收到当前消息，就再也收不到了（消费者就像收音机）
生产者：（可以用作日志收集系统）

文章图片

文章图片
消费者：

文章图片

文章图片
这样，开启多个消费者后，会同时从生产者接收相同的消息。
（2）direct：关键字类型。功能：交换机根据生产者消息中含有的不同的关键字将消息发送给不同的队列，消费者根据不同的关键字从不同的队列取消息
生产者：不用创建对列

文章图片

文章图片
消费者：

文章图片

文章图片
（3）topics：模糊匹配类型。比较常用
发送到一个 topics交换机的消息，它的 routing_key不能是任意的 -- 它的routing_key必须是一个用小数点分割的单词列表。这个字符可以是任何单词，但是通常是一些指定意义的字符。比如：“stock.usd.nyse","nyse.vmw","quick.orange.rabbit".这里可以是你想要路由键的任意字符。最高限制为255字节。生产者与消费者的routing_key必须在同一个表单中。 Topic交换的背后的逻辑类似直接交换（direct） -- 包含特定关键字的消息将会分发到所有匹配的关键字队列中。然后有两个重要的特殊情况：绑定键值： > * (星)可代替一个单词 > # (井) 可代替0个或多个单词生产者：

文章图片

文章图片
消费者：

文章图片

文章图片
4.远程过程调用（RPC）Remote procedure call
消息属性 AMQP协议在一个消息中预先定义了一个包含14个属性的集合。大部分属性很少用到，以下几种除外： > delivery_mode: 标记一个消息为持久的（值为2）或者瞬时的（其它值），你需要记住这个属性（在第二课时用到过） > content_type : 用来描述 MIME 类型的编码 ,比如我们经常使用的 JSON 编码，设置这个属性就非常好实现： application/json > reply_to：reply_to没有特别的意义，只是一个普通的变量名，只是它通常用来命名一个 callback 队列 > correlation_id ：用来关联RPC的请求与应答。关联id的作用：当在一个队列中接收了一个返回，我们并不清楚这个结果时属于哪个请求的，这样当correlation_id属性使用后，我们为每个请求设置一个唯一值，这个值就是关联id。这样，请求会有一个关联id，该请求的返回结果也有一个相同的关联id。然后当我们从callback队列中接收到一个消息后，我们查看一下这个关联，基于这个我们就能将请求和返回进行匹配。如果我们看到一个未知的correlation_id值，我们可以直接丢弃这个消息 -- 它是不属于我们的请求。 RPC执行过程：

文章图片
callback队列我们的RPC将会这样执行： >当客户端启动后，它创建一个匿名的唯一的回调队列 > 对一个RPC请求, 客户端发送一个消息包含两个属性： reply_to （用来设置回调队列）和 correlation_id（用来为每个请求设置一个唯一标识） > 请求发送到 rpc_queue队列 > RPC worker( 服务端) 在那个队列中等待请求，当一个请求出现后，服务端就执行一个job并将结果消息发送给客户端，使用reply_to字段中的队列 > 客户端在callback 队列中等待数据, 当一个消息出现后，检查这个correlation_id属性,如果和请求中的值匹配将返回给应用代码：
rpc_server.py代码 [python] view plain copy print?

#!/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()

服务端代码详单简单： > (4) 和往常一样我们建立一个连接并定义一个队列 > (11) 我们定义了斐波纳契函数，假定输入的都是合法正数 > (19) 我们定义了一个回调的 basic_consume, RPC服务的核心。当收到请求后执行这个函数并返回结果 > (32) 我们可能会执行多个服务端，为了在多个服务端上均匀的分布负荷，我们需要这是 prefetch_count。 rpc_client.py 代码： [python] view plain copy print?

#!/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)

客户端代码稍微复杂些： > (7) 我们建立一个连接，通道并定义一个专门的’callback‘队列用来接收回复 > (16) 我们订阅了“callback”队列，因此我们能够接收 RPC 的返回结果 > (18) ’on_response'在每个返回中执行的回调是一个简单的job，对每个返回消息将检查是否correlation_id使我们需要查找的那个ID，如果是，将保存结果到 self.response 并终端consuming循环 > (23) 下一步，我们定义我们的main方法 - 执行实际的RPC请求 > (24) 在这方法中，首先我们生产一个唯一的 correlatin_id 号并保存 -- 'on_response"回调函数将用着号码来匹配发送和接收的消息值 > (25) 下一步，发布请求信息，使用两个属性： reply_to 和 correlation_id > (32) 这一步我们可以坐等结果的返回 >(33) 最后我们返回结果给用户
或者看下边一篇好理解一点

前面的例子都有个共同点，就是发送端发送消息出去后没有结果返回。如果只是单纯发送消息，当然没有问题了，但是在实际中，常常会需要接收端将收到的消息进行处理之后，返回给发送端。

处理方法描述：发送端在发送信息前，产生一个接收消息的临时队列，该队列用来接收返回的结果。其实在这里接收端、发送端的概念已经比较模糊了，因为发送端也同样要接收消息，接收端同样也要发送消息，所以这里笔者使用另外的示例来演示这一过程。

示例内容：假设有一个控制中心和一个计算节点，控制中心会将一个自然数N发送给计算节点，计算节点将N值加1后，返回给控制中心。这里用center.py模拟控制中心，compute.py模拟计算节点。

compute.py代码分析

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 #!/usr/bin/env python #coding=utf8 import pika #连接rabbitmq服务器 connection = pika.BlockingConnection(pika.ConnectionParameters( host = 'localhost' )) channel = connection.channel() #定义队列 channel.queue_declare(queue = 'compute_queue' ) print ' [*] Waiting for n' #将n值加1 def increase(n): return n + 1 #定义接收到消息的处理方法 def request(ch, method, properties, body): print " [.] increase(%s)" % (body,) response = increase( int (body)) #将计算结果发送回控制中心 ch.basic_publish(exchange = '', routing_key = properties.reply_to, body = str (response)) ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_qos(prefetch_count = 1 ) channel.basic_consume(request, queue = 'compute_queue' ) channel.start_consuming()

计算节点的代码比较简单，值得一提的是，原来的接收方法都是直接将消息打印出来，这边进行了加一的计算，并将结果发送回控制中心。

center.py代码分析

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 #!/usr/bin/env python #coding=utf8 import pika class Center( 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): self .response = body def request( self , n): self .response = None #发送计算请求，并声明返回队列 self .channel.basic_publish(exchange = '', routing_key = 'compute_queue' , properties = pika.BasicProperties( reply_to = self .callback_queue, ), body = str (n)) #接收返回的数据 while self .response is None : self .connection.process_data_events() return int ( self .response) center = Center() print " [x] Requesting increase(30)" response = center.request( 30 ) print " [.] Got %r" % (response,)