一：背景
随着系统的分布式部署，日志的集中处理显得极为重要。本文整合log4j+kafka+storm+mongodb+mysql介绍一下日志的集中处理过程。本方案已应用到线上的正式环境中。

log4j版本：log4j-1.2.17.jar
kafka版本：kafka_2.10-0.8.2.1
zookeeper版本：zookeeper-3.4.7.tar.gz
storm版本：apache-storm-0.9.5.tar.gz
mongodb 版本：mongodb-linux-x86_64-2.6.11.tgz
MySql版本：mysql-5.5.31.tar.gz

二：软件安装
1.kafka安装请参照http://blog.itpub.net/28624388/viewspace-1966761/
2.storm安装请参照http://blog.itpub.net/28624388/viewspace-1973943/
3.mongodb安装请参照http://blog.itpub.net/28624388/viewspace-1974449/
4.mysql安装请参照http://blog.itpub.net/28624388/viewspace-764718/

三：log4j与kafka集成
请参照http://blog.itpub.net/28624388/viewspace-1972027/
日志格式：
logger.warn("处理请求|"+((null==userId || "".equals(userId.toString()))?"0":userId.toString())+"|"+system+"|"+version+"|" + request.getRequestURI() +"|"+GbdDateUtils.format(Calendar.getInstance().getTime(),"yyyy-MM-dd HH:mm:ss")+ "|" + (System.currentTimeMillis() - accessTime) + "|毫秒！");

四：kafka+storm+mongodb+mysql整合
1.代码结构

2.gmap-logs-deal-center模块介绍
该模块实现消费kafka收集到的日志信息，并通过storm简单的处理保存到Mongodb数据库内，通过schedule将mongodb里面的日志数据进行统计分析保存到Mysql数据库内。
a)pom.xml配置文件

点击(此处)折叠或打开

<?xml version="1.0"?>
<project
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.gemdale.gmap</groupId>
<artifactId>GMap</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<groupId>com.gemdale.gmap</groupId>
<artifactId>gmap-logs-deal-center</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>gmap-logs-deal-center</name>
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>gmap-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>gmap-common-logs-bo</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>gmap-common-redis-support</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>18.0</version>
</dependency>
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
</dependency>
<dependency>
<groupId>javax.persistence</groupId>
<artifactId>persistence-api</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-beans</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-mongodb</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>gmap-common-dao-support</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
</dependency>
</dependencies>
</project>

b)Start类

点击(此处)折叠或打开

package com.enjoylink.gmap.logs.deal.center;
import java.util.ArrayList;
import java.util.List;
import org.apache.log4j.Logger;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import com.enjoyslink.gmap.logs.deal.center.stormkafka.StormKafkaBolt;
import com.gemdale.gmap.common.redis.support.RedisUtil;
import com.gemdale.gmap.common.util.ConfigureUtil;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
/**
* @ClassName: Start
* @Description: TODO（用一句话描述这个类的用途）
* @author 信息管理部-gengchong
* @date 2015年10月15日下午5:42:52
*
*/
public class Start {
private static Logger logger = Logger.getLogger(Start.class);
@SuppressWarnings("resource")
public static void main(String[] args) {
// 进行spring的系统初始化化工作,进行必要的数据初始化
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
// 获取接口,并提供服务
// 初始化REDIS服务
logger.info("初始化REDIS服务...");
RedisUtil.initRedisTemplate((RedisTemplate) context.getBean("redisTemplate"));
logger.info("初始化storm-kafka......");
//配置kafka topic
String topic = ConfigureUtil.getProperty("topic", "gmap");
//配置kafka zookeeper
BrokerHosts hosts = new ZkHosts(ConfigureUtil.getProperty("kafka.zk"));
//配置storm spout
SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, "/kafkastorm", "gmapKafka");
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
//配置storm zookeeper
List<String> zkServers = new ArrayList<>();
zkServers.add(ConfigureUtil.getProperty("storm.zk.host"));
spoutConfig.zkServers = zkServers;
spoutConfig.zkPort = Integer.valueOf(ConfigureUtil.getProperty("storm.zk.port"));
spoutConfig.forceFromStart = false;
spoutConfig.socketTimeoutMs = 60 * 1000;
//配置topology
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new KafkaSpout(spoutConfig),
Integer.valueOf(ConfigureUtil.getProperty("storm.max.thread")));
//配置storm bolt
builder.setBolt("bolt",new StormKafkaBolt(), Integer.valueOf(ConfigureUtil.getProperty("storm.max.thread")))
.shuffleGrouping("spout");
//开启topology
Config config = new Config();
config.setDebug(false);
new LocalCluster().submitTopology("topology", config, builder.createTopology());
logger.info("Gmap日志处理中心启动成功！");
}
}

c).StormKafkaBolt类

d).transaction-context.xml

点击(此处)折叠或打开

<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:mongo="http://www.springframework.org/schema/data/mongo" xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.0.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
http://www.springframework.org/schema/data/mongo
http://www.springframework.org/schema/data/mongo/spring-mongo-1.8.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-4.0.xsd">
<bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="location">
<description>datasource config</description>
<value>classpath:context-datasource.properties</value>
</property>
</bean>
<mongo:db-factory host="${mongodb.host}" port="${mongodb.port}" dbname="${mongodb.database}" username="${mongodb.username}" password="${mongodb.password}" />
<bean id="mongoTemplate" class="org.springframework.data.mongodb.core.MongoTemplate">
<constructor-arg name="mongoDbFactory" ref="mongoDbFactory" />
</bean>
<mongo:mapping-converter id="converter" db-factory-ref="mongoDbFactory" />
<bean id="gridFsTemplate" class="org.springframework.data.mongodb.gridfs.GridFsTemplate">
<constructor-arg ref="mongoDbFactory" />
<constructor-arg ref="converter" />
</bean>
<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
<property name="dataSource" ref="platformTomcat" />
</bean>
<bean id="jdbcReadTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
<property name="dataSource" ref="platformReadTomcat" />
</bean>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="platformTomcat"></property>
</bean>
<tx:annotation-driven transaction-manager="transactionManager" proxy-target-class="true"/>
</beans>

e).GmapSystemLogsCollectSchedule类

3.gmap-common-logs-bo代码
GmapSystemLogBo类

ITPub博客

log4j+kafka+storm+mongodb+mysql 日志处理

最新文章