zoukankan      html  css  js  c++  java

  • ActiveMQ学习总结（6）——ActiveMQ集成Spring和Log4j实现异步日志

    我的团队和我正在创建一个由一组RESTful JSON服务组成的服务平台，该平台中的每个服务在平台中的作用就是分别提供一些独特的功能和/或数据。由于平台中产生的日志四散各处，所以我们想，要是能将这些日志集中化处理一下，并提供一个能够让我们查看、过滤、排序和搜索我们所有的日志的基本型的日志查看工具就好了。我们还想让我们的日志是异步式的，因为我们可不想在写日志的时候（比方说，可能会将日志直接写入数据库），让我们提供的服务因为写日志而暂时被阻挡住。

    实现这个目标的策略非常简单明了。

    1. 安装ActiveMQ
    2. 创建一个log4j的日志追加器，将日志写入队列（log4j自带了一个这样的追加器，不过现在让我们自己来写一个吧。） 
    3. 写一个消息侦听器，从MQ服务器上所设置的JMS队列中读取日志并将日志持久化

    下面让我们分步来看这个策略是如何得以实现的。

    安装ActiveMQ

    安装一个外部的ActiveMQ服务器简单极了。这个链接http://servicebus.blogspot.com/2011/02/installing-apache-active-mq-on-ubuntu.html是在Ubuntu上安装ActiveMQ的一个非常棒的指南。你还可以选择在你的应用中嵌入一个消息代理，采用Spring就可以非常轻松实现。 我们将在后文中详谈具体的实现方法。

    创建一个Lo4j的JMS日志追加器

    首先，我们来创建一个log4j的JMS日志追加器。log4j自带了一个这样的追加器（该追加器没有将日志写入一个队列，而是写给了一个话题）

    ?
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.MessageProducer;
    import javax.jms.ObjectMessage;
    import javax.jms.Session;
     
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.log4j.Appender;
    import org.apache.log4j.AppenderSkeleton;
    import org.apache.log4j.Logger;
    import org.apache.log4j.PatternLayout;
    import org.apache.log4j.spi.LoggingEvent;
     
    /**
     * JMSQueue appender is a log4j appender that writes LoggingEvent to a queue.
     * @author faheem
     *
     */
    public class JMSQueueAppender extends AppenderSkeleton implements Appender{
     
    private static Logger logger = Logger.getLogger("JMSQueueAppender");
     
    private String brokerUri;
    private String queueName;
     
    @Override
    public void close() {
     
    }
     
    @Override
    public boolean requiresLayout() {
        return false;
    }
     
    @Override
    protected synchronized void append(LoggingEvent event) {
     
       try {
     
         ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                        this.brokerUri);
     
         // Create a Connection
         javax.jms.Connection connection = connectionFactory.createConnection();
         connection.start();np
     
         // Create a Session
         Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
     
         // Create the destination (Topic or Queue)
         Destination destination = session.createQueue(this.queueName);
     
         // Create a MessageProducer from the Session to the Topic or Queue
         MessageProducer producer = session.createProducer(destination);
         producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
     
         ObjectMessage message = session.createObjectMessage(new LoggingEventWrapper(event));
     
         // Tell the producer to send the message
         producer.send(message);
     
         // Clean up
         session.close();
         connection.close();
      } catch (Exception e) {
         e.printStackTrace();
      }
    }
     
    public void setBrokerUri(String brokerUri) {
        this.brokerUri = brokerUri;
    }
     
    public String getBrokerUri() {
        return brokerUri;
    }
     
    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }
     
    public String getQueueName() {
        return queueName;
    }
    }

    下面让我们看看这里面发生了什么事情。

    第19行：We我们实现了的Log4J日志追加器接口，该接口要求我们实现三个方法：requiresLayout, close和append。我们将暂时简化处理过程，实现所需的append方法。在对logger进行调用时这个方法就会被调用。

    第37行: log4j将一个LoggingEvent对象作为参数对append方法进行调用，这个LoggingEvent对象表示了对logger的一次调用，它封装了每一个日志项的所有信息。

    第41和42行：将指向JMS的uri作为参数，创建一个连接工厂对象，在我们的情况下，该uri指向的是我们的ActiveMQ服务器。

    第45, 46和49行: 我们同JMS服务器建立一个连接和会话。会话有多种打开模式。在Auto_Acknowledge模式的会话中，消息的应答会自动发生。Client_Acknowledge 模式下，客户端需要对消息的接收和/或处理进行显式地应答。另外还有两种其它的模式。有关细节，请参考文档http://download.oracle.com/javaee/1.4/api/javax/jms/Session.html

    第52行: 创建一个队列。将队列的名字作为参数发送给连接

    第56行: 我们将发送模式设置为Non_Persistent。另一个可选的模式是Persistent ，在这种模式下，消息会持久化到一个持久性存储系统中。持久化模式会降低系统速度，但能增加了消息传递的可靠性。

    第58行: 这行我们做了很多事。首先我将一个LoggingEvent对象封装到了一个LoggingEventWrapper对象之中。这么做是因为LoggingEvent对象有一些属性不支持序列化，另外还有一个原因是我想记录一些额外的信息，比如IP地址和主机名。接下来，使用JMS的会话对象，我们把一个对象（LoggingEventWrapper对象）做好了发送前的准备。

    第61行: 我将该对象发送到了队列中。

    下面所示是LoggingEventWrapper的代码。

    ?
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    import java.io.Serializable;
    import java.net.InetAddress;
    import java.net.UnknownHostException;
     
    import org.apache.log4j.EnhancedPatternLayout;
    import org.apache.log4j.spi.LoggingEvent;
     
    /**
     * Logging Event Wraps a log4j LoggingEvent object. Wrapping is required by some information is lost
     * when the LoggingEvent is serialized. The idea is to extract all information required from the LoggingEvent
     * object, place it in the wrapper and then serialize the LoggingEventWrapper. This way all required data remains
     * available to us.
     * @author faheem
     *
     */
     
    public class LoggingEventWrapper implements Serializable{
     
        private static final String ENHANCED_PATTERN_LAYOUT = "%throwable";
        private static final long serialVersionUID = 3281981073249085474L;
        private LoggingEvent loggingEvent;
     
        private Long timeStamp;
        private String level;
        private String logger;
        private String message;
        private String detail;
        private String ipAddress;
        private String hostName;
     
        public LoggingEventWrapper(LoggingEvent loggingEvent){
            this.loggingEvent = loggingEvent;
     
            //Format event and set detail field
            EnhancedPatternLayout layout = new EnhancedPatternLayout();
            layout.setConversionPattern(ENHANCED_PATTERN_LAYOUT);
            this.detail = layout.format(this.loggingEvent);
        }
     
        public Long getTimeStamp() {
            return this.loggingEvent.timeStamp;
        }
     
        public String getLevel() {
            return this.loggingEvent.getLevel().toString();
        }
     
        public String getLogger() {
            return this.loggingEvent.getLoggerName();
        }
     
        public String getMessage() {
            return this.loggingEvent.getRenderedMessage();
        }
     
        public String getDetail() {
            return this.detail;
        }
     
        public LoggingEvent getLoggingEvent() {
            return loggingEvent;
        }
     
        public String getIpAddress() {
            try {
                return InetAddress.getLocalHost().getHostAddress();
            } catch (UnknownHostException e) {
                return "Could not determine IP";
            }
        }
     
        public String getHostName() {
            try {
                return InetAddress.getLocalHost().getHostName();
            } catch (UnknownHostException e) {
                return "Could not determine Host Name";
            }
        }
    }

    消息侦听器

    消息侦听器会对队列（或话题）进行“侦听”。一旦有新消息添加到了队列中，onMessage 方法就会得到调用。

    ?
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.ObjectMessage;
     
    import org.apache.log4j.Logger;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
     
    @Component
    public class LogQueueListener implements MessageListener
    {
        public static Logger logger = Logger.getLogger(LogQueueListener.class);
     
        @Autowired
        private ILoggingService loggingService;
     
        public void onMessage( final Message message )
        {
            if ( message instanceof ObjectMessage )
            {
                try{
                    final LoggingEventWrapper loggingEventWrapper = (LoggingEventWrapper)((ObjectMessage) message).getObject();
                    loggingService.saveLog(loggingEventWrapper);
                }
                catch (final JMSException e)
                {
                    logger.error(e.getMessage(), e);
                } catch (Exception e) {
                logger.error(e.getMessage(),e);
            }
            }
        }
    }

    第23行: 检查从队列中拿到的对象是否是ObjectMessage的实例
    第26行: 从消息中提取出LoggingEventWrapper对象
    第27行: 调用服务方法将日志持久化

    Spring配置

    ?
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    <?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p" xmlns:context="http://www.springframework.org/schema/context" xmlns:jms="http://www.springframework.org/schema/jms" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:aop="http://www.springframework.org/schema/aop" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd">
    <!-- lets create an embedded ActiveMQ Broker -->
    <!-- uncomment the tag below only if you need to create an embedded broker -->
    <!-- amq:broker useJmx="false" persistent="false">
         <amq:transportConnectors>
         <amq:transportConnector uri="tcp://localhost:61616" />
         </amq:transportConnectors>
    </amq:broker-->
    <!-- ActiveMQ destinations to use -->
    <amq:queue id="destination" physicalName="logQueue" />
    <!-- JMS ConnectionFactory to use, configuring the embedded broker using XML -->
    <amq:connectionFactory id="jmsFactory" brokerURL="tcp://localhost:61616" />
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
       <constructor-arg ref="jmsFactory" />
       <property name="exceptionListener" ref="JMSExceptionListener" />
       <property name="sessionCacheSize" value="100" />
    </bean>
    <!-- Spring JMS Template -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
       <constructor-arg ref="connectionFactory" />
    </bean>
    <!-- listener container definition using the jms namespace, concurrency
    is the max number of concurrent listeners that can be started -->
    <jms:listener-container concurrency="10">
       <jms:listener id="QueueListener" destination="logQueue" ref="logQueueListener" />
    </jms:listener-container>
    </beans>

    第5到9行: 使用代理标签建立一个嵌入式消息代理。既然我用的是外部消息代理，所以我就不需要它了。
    第12行: 给出你想要连接的队列的名字
    第14行: 代理服务器的URI
    第15到19行: 连接工厂的设置
    第26到28行: 消息侦听器的设置，这里可以指定用于从队列中读取消息的并发现线程的个数

    当然，上面的例子做不到让你能够拿来就用。你还需要包含所有的JMS依赖库并实现完成日志持久化任务的服务。但是，我希望本文能够为你提供一个相当不错的思路。
  • 相关阅读:
    Java 之Integer相等比较
    Java 之HashMap.values()方法误用
    Java 异常归纳总结
    MySQL之GROUP BY用法误解
    Java 类的成员初始化顺序
    十三、java_GUI
    十二、java_网络编程
    十一、java线程
    十、java_IO
    九、java容器
  • 原文地址：https://www.cnblogs.com/zhanghaiyang/p/7213128.html
Copyright © 2011-2022 走看看