概述
基于Flume + MongoDB，对现有的多个应用系统进行日志采集。
特点
- 采集范围
每一次用户请求的请求信息。 - 数据量大
- 尽量减少现有系统的改动
数据流图
说明：
首先考虑的结构体系，是直接在应用系统中，将日志数据写到Flume；但是现有的应用系统都是非Maven的，需要在每一个应用系统中添加20+个jar包。为避免这种情况，抽出了一层日志服务，开放webservice服务给应用系统调用，最终形成上述的体系。
日志存储
1.需要解决的问题
1.1 借助Flume，写日志到MongoDB
参考：Flume学习应用：Java写日志数据到MongoDB
- 外网参考：Flume学习应用：Java写日志数据到MongoDB
1.2 发布webservice服务
参考：在web项目中发布jaxws
- 外网参考：在web项目中发布jaxws
2.日志服务实现
一个简单的web项目，对外发布一个webservice服务，实现写日志到Flume。
2.1 文件结构
src/main/java|---- cn.sinobest.asj.log|---- ISALog.java # 日志服务接口|---- SALogImpl.java # 日志服务实现类|---- cn.sinobest.asj.log.exception|---- InvalidGradeException.java # 表示无效的日志等级|---- InvalidFormatExceptioin.java # 表示无效的消息格式（要求是JSON格式字符串）|---- cn.sinobest.asj.log.util|---- ValidGrade.java # 枚举，所有有效的日志等级（DEBUG, INFO, WARN, ERROR）|---- MessageTemplate.java # 消息模板 src/main/resources|---- log4j.properties src/main/webapp|---- WEB-INF|---- sun-jaxws.xml|---- web.xml|---- index.jsp # 这个可以忽略 pom.xml
2.2 文件内容
你可以直接从log-service拿到源代码，并跳过这一节的内容。
- pom.xml
pom.xml1 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 2 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> 3 <modelVersion>4.0.0</modelVersion> 4 <groupId>cn.sinobest.asj</groupId> 5 <artifactId>log-service</artifactId> 6 <packaging>war</packaging> 7 <version>0.0.1-SNAPSHOT</version> 8 <name>log-service Maven Webapp</name> 9 <url>http://maven.apache.org</url> 10 <dependencies> 11 <dependency> 12 <groupId>junit</groupId> 13 <artifactId>junit</artifactId> 14 <version>3.8.1</version> 15 <scope>test</scope> 16 </dependency> 17 <dependency> 18 <groupId>log4j</groupId> 19 <artifactId>log4j</artifactId> 20 <version>1.2.16</version> 21 </dependency> 22 <dependency> 23 <groupId>commons-logging</groupId> 24 <artifactId>commons-logging</artifactId> 25 <version>1.1.1</version> 26 </dependency> 27 <!-- for log to Flume --> 28 <dependency> 29 <groupId>org.apache.flume.flume-ng-clients</groupId> 30 <artifactId>flume-ng-log4jappender</artifactId> 31 <version>1.6.0</version> 32 </dependency> 33 <!-- for jax-ws --> 34 <dependency> 35 <groupId>com.sun.xml.ws</groupId> 36 <artifactId>jaxws-rt</artifactId> 37 <version>2.2.10</version> 38 </dependency> 39 <!-- for test the log content is a json-format or not --> 40 <dependency> 41 <groupId>org.mongodb</groupId> 42 <artifactId>mongo-java-driver</artifactId> 43 <version>2.13.0</version> 44 </dependency> 45 </dependencies> 46 <build> 47 <finalName>log-service</finalName> 48 </build> 49 </project>
- web.xml
<?xml version="1.0" encoding="UTF-8"?> <web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns="http://java.sun.com/xml/ns/javaee"xmlns:web="http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"id="WebApp_ID" version="3.0" metadata-complete="false"><display-name>Archetype Created Web Application</display-name> </web-app>
注意：如果是servlet3.0以下的版本，需要额外的配置。
- log4j.properties
# 配置Log4jAppender，能写日志到Flume log4j.appender.flumeAvro=org.apache.flume.clients.log4jappender.Log4jAppender log4j.appender.flumeAvro.Hostname=localhost log4j.appender.flumeAvro.Port=44444 log4j.appender.flumeAvro.UnsafeMode=true log4j.appender.flumeAvro.layout=org.apache.log4j.PatternLayout log4j.appender.flumeAvro.layout.ConversionPattern=%m # set root logger log4j.rootLogger=INFO, flumeAvro
- ISALog.java
package cn.sinobest.asj.log; import javax.jws.WebParam; import javax.jws.WebService; import cn.sinobest.asj.log.exception.InvalidFormatExceptioin; import cn.sinobest.asj.log.exception.InvalidGradeException; /*** SINOBEST ASJ Log - 为实现日志的统一采集和管理.* * @author lijinlong* */ @WebService public interface ISALog {/*** 日志记录.* * @param grade* 日志等级描述 - 忽略大小写.* @param content* 日志内容 - 需要为JSON格式的字符串.*/public void log(@WebParam(name = "grade") String grade,@WebParam(name = "content") String content)throws InvalidGradeException, InvalidFormatExceptioin; }
- SALogImpl.java SALogImpl.java
1 package cn.sinobest.asj.log; 2 import javax.jws.WebService; 3 import org.apache.commons.logging.Log; 4 import org.apache.commons.logging.LogFactory; 5 import cn.sinobest.asj.log.exception.InvalidFormatExceptioin; 6 import cn.sinobest.asj.log.exception.InvalidGradeException; 7 import cn.sinobest.asj.log.util.MessageTemplate; 8 import cn.sinobest.asj.log.util.ValidGrade; 9 import com.mongodb.util.JSON; 10 @WebService(endpointInterface = "cn.sinobest.asj.log.ISALog") 11 public class SALogImpl implements ISALog { 12 static final Log log = LogFactory.getLog(SALogImpl.class); 13 public void log(String grade, String content) throws InvalidGradeException, 14 InvalidFormatExceptioin { 15 checkGrade(grade); 16 checkContent(content); 17 ValidGrade vg = ValidGrade.valueOf(grade.toUpperCase()); 18 log(vg, content); 19 } 20 /** 21 * 根据日志等级，调用{@link log}的不同方法记录日志. 22 * 23 * @param vg 24 * 日志等级 25 * @param content 26 * 日志内容 27 */ 28 private void log(ValidGrade vg, String content) { 29 switch (vg) { 30 case DEBUG: 31 log.debug(content); 32 break; 33 case INFO: 34 log.info(content); 35 break; 36 case WARN: 37 log.warn(content); 38 break; 39 case ERROR: 40 log.error(content); 41 break; 42 default: 43 break; 44 } 45 } 46 /** 47 * 检查日志等级的有效性. 48 * 49 * @param grade 50 * 日志等级描述. 51 * @throws InvalidGradeException 52 * 当日志等级无效时，抛出此异常. 53 */ 54 private void checkGrade(String grade) throws InvalidGradeException { 55 boolean valid = ValidGrade.isValid(grade); 56 if (!valid) { 57 String message = String.format(MessageTemplate.INVALID_GRADE, 58 grade, ValidGrade.getEnumContent()); 59 throw new InvalidGradeException(message); 60 } 61 } 62 /** 63 * 检查日志内容格式的有效性.<br> 64 * 要求为JSON格式的字符串. 65 * 66 * @param content 67 * 日志内容. 68 * @throws InvalidFormatExceptioin 69 * 当日志内容格式无效时，抛出此异常. 70 */ 71 private void checkContent(String content) throws InvalidFormatExceptioin { 72 boolean valid = true; 73 if (content == null || content.isEmpty()) { 74 valid = false; 75 } else { 76 try { 77 JSON.parse(content); 78 valid = true; 79 } catch (com.mongodb.util.JSONParseException e) { 80 valid = false; 81 } 82 } 83 if (!valid) { 84 String message = String.format(MessageTemplate.INVALID_FORMAT, 85 content); 86 throw new InvalidFormatExceptioin(message); 87 } 88 } 89 /** 90 * just for test. 91 * 92 * @param args 93 */ 94 public static void main(String[] args) { 95 String[][] data = { { "info", "{'name':'ljl','age':26}" }, 96 { "INFO", "trouble is a friend." }, 97 { "JOKE", "{'message':'I am feeling down.'}" } }; 98 ISALog ilog = new SALogImpl(); 99 for (String[] dat : data) { 100 String grade = dat[0]; 101 String content = dat[1]; 102 try { 103 ilog.log(grade, content); 104 } catch (Exception e) { 105 e.printStackTrace(); 106 } 107 } 108 } 109 }
- InvalidGradeException.java
package cn.sinobest.asj.log.exception; /*** 表示无效的日志等级.* @author lijinlong**/ public class InvalidGradeException extends Exception {private static final long serialVersionUID = 1341726127995938030L;public InvalidGradeException(String message) {super(message);} }
- InvalidFormatExceptioin.java
package cn.sinobest.asj.log.exception; /*** 表示无效的日志等级.* @author lijinlong**/ public class InvalidGradeException extends Exception {private static final long serialVersionUID = 1341726127995938030L;public InvalidGradeException(String message) {super(message);} }
- ValidGrade.java ValidGrade.java
1 package cn.sinobest.asj.log.util; 2 /** 3 * 有效的日志等级. 4 * 5 * @author lijinlong 6 * 7 */ 8 public enum ValidGrade { 9 DEBUG, INFO, WARN, ERROR; 10 /** 有效日志等级的枚举内容 */ 11 private static String enumContent; 12 /** 13 * 获取所有有效的日志等级. 14 * 15 * @return 16 */ 17 public static String getEnumContent() { 18 if (enumContent != null && !enumContent.isEmpty()) 19 return enumContent; 20 ValidGrade[] vgs = ValidGrade.values(); 21 StringBuilder builder = new StringBuilder(30); 22 for (ValidGrade vg : vgs) { 23 builder.append(vg).append(","); 24 } 25 builder.delete(builder.length() - 1, builder.length()); 26 enumContent = builder.toString(); 27 return enumContent; 28 } 29 30 /** 31 * 判断日志等级是否有效. 32 * @param grade 日志等级 - 忽略大小写. 33 * @return 34 */ 35 public static boolean isValid(String grade) { 36 if (grade == null || grade.isEmpty()) 37 return false; 38 39 boolean result = false; 40 41 final String GRADE = grade.toUpperCase(); 42 ValidGrade[] vgs = ValidGrade.values(); 43 for (ValidGrade vg : vgs) { 44 if (vg.toString().equals(GRADE)) { 45 result = true; 46 break; 47 } 48 } 49 50 return result; 51 } 52 53 /** 54 * just for test. 55 * @param args 56 */ 57 public static void main(String[] args) { 58 String content = getEnumContent(); 59 System.out.println(content); 60 61 String[] testGrade = {"DEBUG", "INFO", "WARN", "ERROR", "TEST"}; 62 for (String tg : testGrade) { 63 if (!ValidGrade.isValid(tg)) { 64 String message = String.format("%s is invalid.", tg); 65 System.out.println(message); 66 } 67 } 68 } 69 }
- MessageTemplate.java
package cn.sinobest.asj.log.util; /*** 消息模板.* @author lijinlong**/ public class MessageTemplate {/** 无效的消息等级 */public static final String INVALID_GRADE = "无效的日志等级[%s]。服务支持的日志等级有：%s。";/** 无效的消息内容格式 */public static final String INVALID_FORMAT = "无效的日志内容格式:\n%s\n，请检查是否为JSON格式的字符串。"; }
- sun-jaxws.xml
<?xml version="1.0" encoding="UTF-8"?> <endpoints xmlns="http://java.sun.com/xml/ns/jax-ws/ri/runtime"version="2.0"><endpoint name="defaultLog" implementation="cn.sinobest.asj.log.SALogImpl"url-pattern="/log.action" /> </endpoints>
应用系统群
1.需要考虑的问题
1.1 拦截
使用Filter可以实现拦截。
1.2 组织日志内容
视需求而定，当前仅对request中的部分信息进行了采集。
1.3 格式化
日志信息需要格式化为JSON字符串，才能正确的写到MongoDB。
1.4 请求webservice服务
参考：基于wsimport生成代码的客户端
- 外网参考：基于wsimport生成代码的客户端
2. demo
2.1 文件结构图
src|---- cn.sinobest.asj.log|---- LogFilter.java|---- cn.sinobest.asj.log.wsimport # 存放wsimport生成的代码# 省略 basic|---- WEB-INF|---- web.xml
2.2 文件内容
- LogFilter.java LogFilter.java
1 package cn.sinobest.asj.log; 2 import java.io.IOException; 3 import java.net.MalformedURLException; 4 import java.net.URL; 5 import java.util.Date; 6 import java.util.HashMap; 7 import java.util.Map; 8 import javax.servlet.Filter; 9 import javax.servlet.FilterChain; 10 import javax.servlet.FilterConfig; 11 import javax.servlet.ServletException; 12 import javax.servlet.ServletRequest; 13 import javax.servlet.ServletResponse; 14 import javax.servlet.http.HttpServletRequest; 15 import org.apache.commons.logging.Log; 16 import org.apache.commons.logging.LogFactory; 17 import org.json.JSONObject; 18 import cn.sinobest.asj.log.wsimport.ISALog; 19 import cn.sinobest.asj.log.wsimport.InvalidFormatExceptioin_Exception; 20 import cn.sinobest.asj.log.wsimport.InvalidGradeException_Exception; 21 import cn.sinobest.asj.log.wsimport.SALogImplService; 22 public class LogFilter implements Filter { 23 static final Log log = LogFactory.getLog(LogFilter.class); 24 static final String WSDL_LOCATION = "http://localhost:8080/logserv/log.action?wsdl"; 25 @Override 26 public void destroy() { 27 } 28 @Override 29 public void doFilter(ServletRequest request, ServletResponse response, 30 FilterChain chain) throws IOException, ServletException { 31 try { 32 log(request); 33 } catch (InvalidFormatExceptioin_Exception e) { 34 e.printStackTrace(); 35 } catch (InvalidGradeException_Exception e) { 36 e.printStackTrace(); 37 } finally { 38 chain.doFilter(request, response); 39 } 40 } 41 private void log(ServletRequest request) throws MalformedURLException, 42 InvalidFormatExceptioin_Exception, InvalidGradeException_Exception { 43 Map<String, Object> data = new HashMap<String, Object>(); 44 data.put("appid", "zfba"); 45 data.put("time", new Date()); 46 data.put("localAddr", request.getLocalAddr()); 47 data.put("localName", request.getLocalName()); 48 data.put("localPort", request.getLocalPort()); 49 data.put("remoteAddr", request.getRemoteAddr()); 50 data.put("remoteHost", request.getRemoteHost()); 51 data.put("remotePort", request.getRemotePort()); 52 // data.put("serverName", request.getServerName()); 53 // data.put("serverPort", request.getServerPort()); 54 HttpServletRequest hrequest = (HttpServletRequest) request; 55 data.put("pathInfo", hrequest.getPathInfo()); 56 data.put("pathTranslated", hrequest.getPathTranslated()); 57 data.put("remoteUser", hrequest.getRemoteUser()); 58 data.put("requestURI", hrequest.getRequestURI()); 59 data.put("requestURL", hrequest.getRequestURL()); 60 data.put("servletPath", hrequest.getServletPath()); 61 JSONObject cont = new JSONObject(data); 62 URL url = new URL(WSDL_LOCATION); 63 SALogImplService ss = new SALogImplService(url); 64 ISALog service = ss.getSALogImplPort(); 65 service.log("info", cont.toString()); 66 } 67 @Override 68 public void init(FilterConfig arg0) throws ServletException { 69 } 70 }
- web.xml
这里仅贴出新增的内容：<!-- 测试日志 --><filter><filter-name>log-filter</filter-name><filter-class>cn.sinobest.asj.log.LogFilter</filter-class></filter><!-- 测试日志 --><filter-mapping><filter-name>log-filter</filter-name><url-pattern>/*</url-pattern></filter-mapping>
测试
- 启动MongoDB
参考《Flume学习应用：Java写日志数据到MongoDB》 - 配置并启动Flume
参考《Flume学习应用：Java写日志数据到MongoDB》 - 启动日志服务
参考《在web项目中发布jaxws》 - 启动应用系统，并进行访问
- 查看MongoDB数据库
参考《Flume学习应用：Java写日志数据到MongoDB》
附录
相关文章
- Flume学习应用：Java写日志数据到MongoDB
博客园：Flume学习应用：Java写日志数据到MongoDB - 在web项目中发布jaxws
博客园：在web项目中发布jaxws - 基于wsimport生成代码的客户端
博客园：基于wsimport生成代码的客户端
