RabbitMQ trace 日志调试 - 为程序员服务
RabbitMQ trace 日志调试
RabbitMQ 默认日志里只有类似客户端“accpet/close”等信息，对于有异常或者跟踪消息内部结构就比较麻烦了。
翻阅官方教程意外发现了 rabbitmq_tracing 插件和 。
注意：打开 trace 会影响消息写入功能，适当打开后请关闭。
自己顺手写了一个封装脚本，参考：
安装上面的插件并开启 trace_on 之后，会发现多了两个 exchange：amq.rabbitmq.trace 和 amq.rabbitmq.log，类型均为：topic。
懂了吧，只要订阅这两个主题，就能收到：客户端连接、消息发收等具体信息了。
下面开始测试吧，先安装插件，并打开 trace_on：
$ sudo rabbitmq-plugins enable rabbitmq_tracing
$ （略去重启RabbitMQ命令）
$ sudo rabbitmqctl trace_on
Starting tracing for vhost "/" ...
测试代码采用 Python 的 pika 库，片段如下：
import pika
def _on_message(ch, method, properties, body):
ret['routing_key'] = method.routing_key
ret['headers'] = properties.headers
ret['body'] = body
conn = pika.BlockingConnection(pika.ConnectionParameters())
chan.queue_declare(exclusive=False, auto_delete=True) # 临时队列
queue = ret.method.queue
chan.queue_bind(exchange='amq.rabbitmq.trace', queue=queue, routing_key='#')
chan.queue_bind(exchange='amq.rabbitmq.log', queue=queue, routing_key='#')
chan.basic_consume(_on_message, queue=queue, no_ack=True)
chan.start_consuming()
运行结果大致如下：
{'body': 'accepting AMQP connection &0.28967.1& (127.0.0.1:52930 -& 127.0.0.1:5672)\n',
'headers': None, 'routing_key': 'info'}
{'body': 'accepting AMQP connection &0.28967.1& (127.0.0.1:52930 -& 127.0.0.1:5672)\n',
'headers': {'node': 'rabbit@mac', 'exchange_name ': 'amq.rabbitmq.log', 'redelivered': 0,
'routing_keys': ['info'], 'properties': {'timestamp': ,
'content_type': 'text/plain'}},
'routing_key': 'deliver.amq.gen-dJNnBwGUuigsfAzoUD9Zlw'}
{'body': '内容略去...', 'headers': {'node': 'rabbit@mac', 'exchange_name': 'amq.direct',
'routing_keys': ['test2'], 'properties': {}}, 'routing_key': 'publish.amq.direct'}
{'body': 'closing AMQP connection &0.28967.1& (127.0.0.1:52930 -& 127.0.0.1:5672)\n',
'headers': None, 'routing_key': 'info'}
{'body': 'closing AMQP connection &0.28967.1& (127.0.0.1:52930 -& 127.0.0.1:5672)\n',
'headers': {'node': 'rabbit@mac', 'exchange_name': 'amq.rabbitmq.log', 'redelivered': 0,
'routing_keys': ['info'], 'properties': {'timestamp': ,
'content_type': 'text/plain'}},
'routing_key': 'deliver.amq.gen-dJNnBwGUuigsfAzoUD9Zlw'}
原文地址：, 感谢原作者分享。
您可能感兴趣的代码1781人阅读
rabbitMQ中consumer通过建立到queue的连接，创建channel对象，通过channel通道获取message,
Consumer可以声明式的以API轮询poll的方式主动从queue的获取消息，也可以通过订阅的方式被动的从Queue中消费消息,
最近翻阅了基于java的客户端的相关源码，简单做个分析。
编程模型伪代码如下：
ConnectionFactory factory = new ConnectionFactory();
Connection conn = factory.newConnection();
Channel channel=conn.createChannel();
创建Connection需要指定MQ的物理地址和端口，是socket tcp物理连接，而channel是一个逻辑的概念，支持在tcp连接上创建多个MQ channel
以下是基于channel上的两种消费方式。
1、Subscribe订阅方式
boolean autoAck =
channel.basicConsume(queueName, autoAck, &myConsumerTag&,
new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException
String routingKey = envelope.getRoutingKey();
String contentType = properties.contentT
long deliveryTag = envelope.getDeliveryTag();
// (process the message components here ...)
channel.basicAck(deliveryTag, false);
订阅方式其实是向queue注册consumer,通过rpc向queue server发送注册consumer的消息，rabbitMQ Server在收到消息后，根据消息的内容类型判断这是一个订阅消息，
这样当MQ 中queue有消息时，会自动把消息通过该socket(长连接)通道发送出去。
参见ChannelN中的方法&&&
public String basicConsume(String queue, boolean autoAck, String consumerTag,
boolean noLocal, boolean exclusive, Map&String, Object& arguments,
final Consumer callback)
throws IOException
rpc((Method)
new Basic.Consume.Builder()
.queue(queue)
.consumerTag(consumerTag)
.noLocal(noLocal)
.noAck(autoAck)
.exclusive(exclusive)
.arguments(arguments)
return k.getReply();
} catch(ShutdownSignalException ex) {
throw wrap(ex);
Consumer接收消息的过程:
创建Connection后，会启动MainLoop后台线程，循环从socket(FrameHandler)中获取数据包(Frame)，调用channel.handleFrame(Frame frame)处理消息，
&&& public void handleFrame(Frame frame) throws IOException {
&&&&&&& AMQCommand command = _
&&&&&&& if (command.handleFrame(frame)) { // 对消息进行协议assemble
&&&&&&&&&&& _command = new AMQCommand(); // prepare for the next one
&&&&&&&&&&& handleCompleteInboundCommand(command);//对消息消费处理
ChannelN.handleCompleteInboundCommand
&&&&&& ---ChannelN.processAsync
&&&&&&&&&& ----dispatcher.handleDelivery
&&&&&&&&&&&&&&& &---QueueingConsumer.handleDelivery
&&&&&&&&&&&&&&&&&&&& ---this._queue.add(new Delivery(envelope, properties, body));//消息最终放到队列中
每个Consumer都有一个BlockQueue，用于缓存从socket中获取的消息。
接下来，Consumer对象就可以调用api来从客户端缓存的_queue中依次获取消息，进行消费，参见QueueingConsumer.nextDelivery()
对于这种长连接的方式，没看到心跳功能，以防止长连接的因网络等原因连接失效
2、poll API方式
ChannelN：
GetResponse basicGet(String queue, boolean autoAck)
这种方式比较简单，直接通过RPC从MQ Server端获取队列中的消息
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：118634次
积分：2733
积分：2733
排名：第18617名
原创：115篇
转载：255篇
(7)(11)(4)(8)(6)(3)(2)(1)(8)(6)(17)(41)(11)(15)(6)(11)(12)(19)(20)(11)(22)(7)(1)(43)(37)(24)(1)(2)(1)(1)(1)(5)(4)(1)