監聽RabbitMQ日誌
阿新 • • 發佈:2019-01-04
win下,檢視rabbitmq的exchange:.\rabbitmqctl list_exchanges,可以看到有個amq.rabbitmq.log的topic型別的交換器,這個就是mq的日誌輸出的exchange
1、建立連線、channel
ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
2、宣告這個exchange,在vhost下的這個exchange的durable:true,auto_delete:false,internal:true,宣告時如果引數型別不匹配,會報錯
channel.exchangeDeclare("amq.rabbitmq.log", "topic", true, false,true,null);
3、宣告三個日誌等級的訊息佇列,佇列名隨便取
channel.queueDeclare("info_queue", false, false, false, null);
channel.queueDeclare("error_queue", false, false, false, null);
channel.queueDeclare("warning_queue", false, false, false, null);
4、將佇列繫結到交換器上(RabbitMQ把日誌傳送到這個交換器上,並且是以日誌等級為route_key:error、warning、info)
channel.queueBind("info_queue","amq.rabbitmq.log","info");
channel.queueBind("error_queue","amq.rabbitmq.log","error");
channel.queueBind("warning_queue","amq.rabbitmq.log","warning");
5、建立消費者,並設計訊息處理邏輯(這裡僅做列印)
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)throws IOException{
String message = new String(body,"UTF-8");
LOGGER.info("-----------MQ LOG-------------\n"+ message+"\n\n");
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
6、指定消費佇列,監聽佇列並消費
channel.basicConsume("info_queue", true, consumer);
channel.basicConsume("error_queue", true, consumer);
channel.basicConsume("warning_queue", true, consumer);
7、控制檯的顯示如下:
完成的程式碼:
package com.rabbitmq.ex.rabiitmqone.service;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author jimsTon
* @date 2018/5/24
*/
public class MqLogListenService {
private static final Logger LOGGER = LoggerFactory.getLogger(MqLogListenService.class);
/**
* rabbitmq日誌訊息接收的交換器
*/
private static final String EXCHANGE_NAME = "amq.rabbitmq.log";
/**
* 自定義info級別的佇列
*/
private static final String INFO_QUEUE = "info_queue";
/**
* 自定義warning級別的佇列
*/
private static final String WARNING_QUEUE = "warning_queue";
/**
* 自定義error級別的佇列
*/
private static final String ERROR_QUEUE = "error_queue";
private void receiveMqLog() throws IOException, TimeoutException {
//建立channel
ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//宣告交換器
channel.exchangeDeclare(EXCHANGE_NAME, "topic", true, false,true,null);
//宣告佇列
channel.queueDeclare(INFO_QUEUE, false, false, false, null);
channel.queueDeclare(ERROR_QUEUE, false, false, false, null);
channel.queueDeclare(WARNING_QUEUE, false, false, false, null);
//將佇列與交換器繫結,並指定route_key,rabbit預設的等級日誌傳送路由鍵是info、warning、error
channel.queueBind(INFO_QUEUE,EXCHANGE_NAME,"info");
channel.queueBind(ERROR_QUEUE,EXCHANGE_NAME,"error");
channel.queueBind(WARNING_QUEUE,EXCHANGE_NAME,"warning");
//建立消費者
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)throws IOException{
String message = new String(body,"UTF-8");
LOGGER.info("-----------MQ LOG-------------\n{}",message);
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
channel.basicConsume(INFO_QUEUE, true, consumer);
channel.basicConsume(ERROR_QUEUE, true, consumer);
channel.basicConsume(WARNING_QUEUE, true, consumer);
}
public static void main(String []args) throws IOException, TimeoutException {
MqLogListenService mqLogListenService = new MqLogListenService();
mqLogListenService.receiveMqLog();
}
}