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

2022-08-04 15:50:31

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

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

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

  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依赖库并实现完成日志持久化任务的服务。但是，我希望本文能够为你提供一个相当不错的思路。

上一篇：重装Ubuntu16.04及安装theano


下一篇：关于Web安全的链接文章