ActiveMQ自定义日志插件

刺骨的言语ヽ痛彻心扉 2022-11-20 08:13 916阅读 0赞

在使用mq时，对消息何时发送，是否发送成功，消息是否消费成功做一个记录，activeMQ自身有一个日志插件LoggingBrokerPlugin，可以记录一些事件：

但在记录消息发送事件时，对消息内容的输出不够直观，如下图，消息内容为content = org.apache.activemq.util.ByteSequence@6a3207aa：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NodWl4dWUyNA_size_16_color_FFFFFF_t_70][]

为了显示更直观的消息发送内容，需要写一个自定义的日志插件，根据ActiveMQ的插件机制，需要编写两个类：

CustomSendMsgPlugin：

import org.apache.activemq.broker.Broker;
    import org.apache.activemq.broker.BrokerPlugin;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class CustomSendMsgPlugin implements BrokerPlugin{
        private static final Logger LOG = LoggerFactory.getLogger(CustomSendMsgPlugin.class);
    
    	@Override
    	public Broker installPlugin(Broker broker) throws Exception {
    		LOG.info("--安装自定义发送消息插件--");   
    		 return new CustomSendMsgLog(broker); 
    	}
    
    }

CustomSendMsgLog：

import java.io.UnsupportedEncodingException;
    
    import org.apache.activemq.broker.Broker;
    import org.apache.activemq.broker.BrokerFilter;
    import org.apache.activemq.broker.ConsumerBrokerExchange;
    import org.apache.activemq.broker.ProducerBrokerExchange;
    import org.apache.activemq.command.ActiveMQDestination;
    import org.apache.activemq.command.Message;
    import org.apache.activemq.command.MessageAck;
    import org.apache.activemq.util.ByteSequence;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class CustomSendMsgLog extends BrokerFilter {
    	private static final Logger LOG = LoggerFactory.getLogger(CustomSendMsgLog.class);
    
    	public CustomSendMsgLog(Broker next) {
    		super(next);
    		LOG.info("初始化发送消息过滤器");
    	}
    
    	@Override
    	public void send(ProducerBrokerExchange producerExchange, Message messageSend) throws Exception {
    		logSend(messageSend.copy());
    		super.send(producerExchange, messageSend);
    	}
    
    	private void logSend(Message copy) {
    		copy.getSize();
    		Logger perDestinationsLogger = LOG;
    		ActiveMQDestination destination = copy.getDestination();
    		perDestinationsLogger = LoggerFactory.getLogger(
    				LOG.getName() + "." + destination.getDestinationTypeAsString() + "." + destination.getPhysicalName());
    		ByteSequence bs = copy.getContent();
    		String content = null;
    		try {
    			content = new String(bs.getData(), "utf-8");
    		} catch (UnsupportedEncodingException e) {
    			e.printStackTrace();
    		}
    
    		perDestinationsLogger.info("Sending message: {} ActiveMQStringMessage {}", copy,content);
    	}
    	
    	@Override
        public void acknowledge(ConsumerBrokerExchange consumerExchange, MessageAck ack) throws Exception {
                LOG.info("Acknowledging message for client ID: {}{}", consumerExchange.getConnectionContext().getClientId(), (ack.getMessageCount() == 1 ? ", " + ack.getLastMessageId() : ""));
                if (ack.getMessageCount() > 1) {
                    LOG.trace("Message count: {}, First Message Id: {}, Last Message Id: {}", new Object[]{ ack.getMessageCount(), ack.getFirstMessageId(), ack.getLastMessageId() });
                }
            super.acknowledge(consumerExchange, ack);
        }
    }

然后在服务器端配置文件active.xml中配置：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NodWl4dWUyNA_size_16_color_FFFFFF_t_70 1][]重启mq后，查看日志：

![20201027163643428.png][]

消息内容已可以直观的看到。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NodWl4dWUyNA_size_16_color_FFFFFF_t_70]: /images/20221120/de9d981bd181466eb4130c3b6516e2b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NodWl4dWUyNA_size_16_color_FFFFFF_t_70 1]: /images/20221120/d9db2995c07e4fdabc15114ab1ca9aff.png
[20201027163643428.png]: /images/20221120/7351f9d1e6b54650b4860a0a0c3a991a.png