監聽RabbitMQ日誌

win下，查看rabbitmq的exchange：.\rabbitmqctl list_exchanges，能夠看到有個amq.rabbitmq.log的topic類型的交換器，這個就是mq的日誌輸出的exchange

一、建立鏈接、channel

ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

二、聲明這個exchange，在vhost下的這個exchange的durable：true，auto_delete：false，internal：true,聲明時若是參數類型不匹配，會報錯

channel.exchangeDeclare("amq.rabbitmq.log", "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);

四、將隊列綁定到交換器上（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");

五、建立消費者，並設計消息處理邏輯（這裏僅作打印）

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);
            }

        };

六、指定消費隊列，監聽隊列並消費

channel.basicConsume("info_queue", true, consumer);
channel.basicConsume("error_queue", true, consumer);
channel.basicConsume("warning_queue", true, consumer);

七、控制檯的顯示以下：

 

完成的代碼：

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();
    }
}