摘要:在发送后端监听声明的排他队列,当收到消息后比对正确则处理消息断开监听连接,然后此队列被系统自动回收。并且通过也看到了这条消息的返回。此时我们基本已经将问题锁定在端了。
背景
公司的一个项目使用rabbitmq作为broker进行交互,并且数据的查询方法使用RPC模式,RPC Client端使用java编写并使用springAMQP包与rabbitmq交互,在RPC Server端使用python的 pika包与rabbitmq交互。两端都使用标准官方例程,发现在Client端发送的消息可以被Server端接收并处理然后返回结果,但是Client端只会会收到一个null值。
问题排查 1 理解传统的RPC模式运行流程传统模式下 Client端向一个指定的队列里推送消息,并声明一个一次性排他队列,然后将发送消息头部的reply-to属性的值设置为队列的名字,correlation_id属性设置为一个随机生成的值用于消息鉴定然后发送消息。在发送后Client端监听声明的排他队列,当收到消息后比对correaltiion_id,正确则处理消息断开监听连接,然后此队列被系统自动回收。 在Server端收到消息后处理消息然后将消息返回,返回的消息的routing-key设置为reply-to的值,properties中设置correlation_id为收到的correlation_id值。这样就完成一次RPC交互模式。
要解决今天这个问题我们还要知道几个知识点:
1当消息发送到exchange后如果没有队列接收此消息,那么此消息就会丢失。
2 一次性的排他队列在Client不在监听此队列就会自动被rabbitmq删除。
排查1 Client端收到的Null值从哪里来?因为我是使用python写RPC Server端并且我也不怎么会java代码。……
所以这个null值从那里来我就无法从Client端下手。那我们只能从Server端进行排查。(最后我认为是在java代码编写错误(是自己的代码)的情况下 springAMQP返回的一个默认值)
在Server端打印收到的message并打印此消息的header信息和body信息,看到在reply-to中就是Client端设置的队列。并且通过rabbitmq也看到了这条消息的返回。
排查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.
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 官方介绍
难点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代码
# -*- 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,或者官方文档,官方论坛。我个人认为官方文档真的是好东西。无数的浅坑的解决办法都在官方文档。当然深坑就不说了那就是论坛加能力加运气才能排查出来的了。不过官方大多都是英文。真是愁人,我 加强英语能力吧。
文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。
转载请注明本文地址:https://www.ucloud.cn/yun/40679.html
一、RabbitMQ整合SpringAMQP RabbitAdmin RabbitAdmin类可以很好的操作RabbitMQ,在Spring中直接进行注入即可。 注意:autoStartup必须要设置为true,否则Spring容器不会加载RabbitAdmin类 RabbitAdmin底层实现就是从Spring容器中获取Exchange、Binding、RoutingKey以及Queue的...
摘要:后续介绍交换机,生产者直接将消息投递到中。消息,服务器和应用程序之间传送的数据,由和组成。也称为消息队列,保存消息并将它们转发给消费者。主要是应为和有一个绑定的关系。 showImg(https://img-blog.csdnimg.cn/20190509221741422.gif); showImg(https://img-blog.csdnimg.cn/20190731191914...
摘要:为了避免与参数混淆,我们将其称为绑定键。直接交换我们之前教程的日志记录系统将所有消息广播给所有消费者。在这种设置中,使用路由键发布到交换机的消息将被路由到队列。所有其他消息将被丢弃。 源码:https://github.com/ltoddy/rabbitmq-tutorial 路由 本章节教程重点介绍的内容 在之前的教程中,我们构建了一个简单的日志系统 我们能够将日志消息广播给许多接收...
摘要:需要特别明确的概念交换机的持久化,并不等于消息的持久化。消息的处理,是有两种方式,一次性。在上述示例中,使用的,意味着接收全部的消息。注意与是两个不同的队列。后端处理,可以针对每一个启动一个或多个,以提高消息处理的实时性。 RabbitMQ与PHP(一) 项目中使用RabbitMQ作为队列处理用户消息通知,消息由前端PHP代码产生,处理消息使用Python,这就导致代码一致性问题,调...
摘要:为了预防消息丢失,提供了,即工作进程在收到消息并处理后,发送给,告知这时候可以把该消息从队列中删除了。如果工作进程挂掉了,没有收到,那么会把该消息重新分发给其他工作进程。之前在发布消息时,的值为即使用。 HelloWorld 简介 RabbitMQ:接受消息再传递消息,可以视为一个邮局。发送者和接受者通过队列来进行交互,队列的大小可以视为无限的,多个发送者可以发生给一个队列,多个接收者...
阅读 2550·2019-08-30 10:53
阅读 3131·2019-08-29 16:20
阅读 2898·2019-08-29 15:35
阅读 1712·2019-08-29 12:24
阅读 2821·2019-08-28 18:19
阅读 1812·2019-08-23 18:07
阅读 2252·2019-08-23 15:31
阅读 1113·2019-08-23 14:05