rabbitmq direct reply-to 在springAMQP和python之间的使用

时间 2019-11-11

标签 rabbitmq direct reply springamqp python 之间使用栏目 RabbitMQ 繁體版

原文原文链接

背景

公司的一个项目使用rabbitmq做为broker进行交互，而且数据的查询方法使用RPC模式，RPC Client端使用java编写并使用springAMQP包与rabbitmq交互，在RPC Server端使用python的 pika包与rabbitmq交互。两端都使用标准官方例程，发如今Client端发送的消息能够被Server端接收并处理而后返回结果，可是Client端只会会收到一个null值。html

问题排查

1 理解传统的RPC模式运行流程

传统模式下 Client端向一个指定的队列里推送消息，并声明一个一次性排他队列，而后将发送消息头部的reply-to属性的值设置为队列的名字，correlation_id属性设置为一个随机生成的值用于消息鉴定而后发送消息。在发送后Client端监听声明的排他队列，当收到消息后比对correaltiion_id,正确则处理消息断开监听链接，而后此队列被系统自动回收。在Server端收到消息后处理消息而后将消息返回，返回的消息的routing-key设置为reply-to的值，properties中设置correlation_id为收到的correlation_id值。这样就完成一次RPC交互模式。
要解决今天这个问题咱们还要知道几个知识点：java

1当消息发送到exchange后若是没有队列接收此消息，那么此消息就会丢失。
2 一次性的排他队列在Client不在监听此队列就会自动被rabbitmq删除。

排查1 Client端收到的Null值从哪里来？

由于我是使用python写RPC Server端而且我也不怎么会java代码。……
因此这个null值从那里来我就没法从Client端下手。那咱们只能从Server端进行排查。（最后我认为是在java代码编写错误（是本身的代码）的状况下 springAMQP返回的一个默认值）python

排查2 Server端收到消息后是否正确的将消息返回

在Server端打印收到的message并打印此消息的header信息和body信息，看到在reply-to中就是Client端设置的队列。而且经过rabbitmq也看到了这条消息的返回。spring

排查3 观察消息有没有被推送回reply-to队列

而后我在Server端收到消息后的callback函数的头部大了断点，接收到消息后Server端程序挂起。此时我去查看reply-to中的队列，发现其已经不存在于rabbitmq中了。由上面的传统RPC模式我推断出多是Client端发送代码后没有监听reply-to队列形成队列消失，而后Server端发送的消息由于没有接收队列而被丢弃。此时咱们基本已经将问题锁定在Client端了。可是Client端的代码是按照rabbitmq官方给的例程书写，应该是没有问题的。此时彷佛陷入了僵局。服务器

定位问题：Google大发加官方文档

这时候我Google一下SpringAMQP框架的是如何写RPC代码？在一些帖子中我发现有的代码会添加一个Listener的类，但有的又不添加。咱们假设他们都是能够运行的。那么是什么缘由会形成这种状况呢？我第一个就是想到了版本问题。随着版本的改变可能代码也会发生变化。以后我就在SpringAMQP的官方文档里面进行查找。果真被我找到了，官方文档里面有这样一段描述：框架

Starting with version 3.4.0, the RabbitMQ server now supports Direct reply-to; this eliminates the main reason for a fixed reply queue (to avoid the need to create a temporary queue for each request). Starting with Spring AMQP version 1.4.1 Direct reply-to will be used by default (if supported by the server) instead of creating temporary reply queues. When no replyQueue is provided (or it is set with the name amq.rabbitmq.reply-to), the RabbitTemplate will automatically detect whether Direct reply-to is supported and either use it or fall back to using a temporary reply queue. When using Direct reply-to, a reply-listener is not required and should not be configured.ide

springAMQP官方地址
翻译一下大致意思就是在RabbitMQ3.4.0版本之后官方提供一种叫作Direct reply-to的方式来实现RPC（这种方式能够极大的提升RPC的性能，由于他不须要每次请求Client端都要新申请一个队列，以后我会再写一篇来详细介绍（翻译 o(∩_∩)o 哈哈）这个特性。而且在SpringAMQP version 1.4.1版本以后默认使用特性，看了一下服务器上的rabbitmq版本3.3.0 这个真的老果真不支持，SpringAMQP的版本果真也是高于这个版本，问题找到。开心，可是怎么解决呢？
Direct reply-to 官方介绍函数

解决方案

一：提高rabbitmq版本，并使两端代码适配direct reply-to 方式

难点1 python的官网没有给例程，不过给了介绍也告诉了如何来实现
难点2 服务器提高版本，已经有业务跑在上面了，我这种对rabbitmq的萌新对rabbitmq各版本升级后的改变并非很了解，估计是难说动领导换了。

针对难点2 我就不想了不过难点1的我已经写出来python如何适配direct reply-to的代码。
更改都是在Client端，Server端仍是能够保持不变。主要主机这几个方面性能

1 reply-to的名字更改成‘amq.rabbitmq.reply-to’这条虚拟队列，你在rabbitmq的控制台上是看不到这条队列的。
2 而后Client监听这条队列的时候要设为为no-ack模式。

下面是根据官方python RPC代码更改的适配 Direct reply-to的python代码
Client端 python代码fetch

# -*- coding:utf-8 -*-  
#!/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)
        # 监听队列为 amp.rabbitmq.reply-to 启动no_ack 模式
        self.channel.basic_consume(self.on_response,
                                   queue='amq.rabbitmq.reply-to',
                                   no_ack=True)

    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,
                                         # 更改了队列名字
                                         reply_to='amq.rabbitmq.reply-to',
                                         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)

Server端代码没有改动

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

解决办法2 java代码不使用默认的direct reply-to模式

这个办法由于我不是写java的因此我只能写一些我在官方文档里面理解的东西了。就是当你不使用SpringAMQP的默认RPC模式的化须要增长Listener对象来监听本身的队列。

RabbitTemplate rabbitTempete=new RabbitTemplate(connectionFactory);  
            rabbitTempete.setExchange(exchangeName);  
            rabbitTempete.setRoutingKey(topic);  
           //比官方文档多的
            Queue  replyqQueue=replyQueue();  
            admin.declareQueue(replyqQueue); 
            rabbitTempete.setReplyQueue(replyqQueue);  
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();  
            container.setConnectionFactory(connectionFactory);  
            container.setQueues(replyqQueue);  
            container.setMessageListener(rabbitTempete);  
            container.start();  
            //比官方文档多的中止
            Object  response=rabbitTempete.convertSendAndReceive(t);

SpringAMQP书写官方文档
相比较要本身申请队列本身监听。不过我也没试过这段代码就不知道能不能用了。

总结

这个问题基本获得很好的解决了。解决一个问题首先你要明白一个东西正常状况下是一种什么情况，而后出了问题就从前日后，从后往前，从中往两边等等等。而后Google，或者官方文档，官方论坛。我我的认为官方文档真的是好东西。无数的浅坑的解决办法都在官方文档。固然深坑就不说了那就是论坛加能力加运气才能排查出来的了。不过官方大多都是英文。真是愁人，我增强英语能力吧。