关于后台行为记录，可理解为对用户请求的监控日志。也可理解为埋点的一种。通常埋点可简单分为前端埋点和后端埋点。
前端埋点需要对APP或Web的一些控件进行有代码或无代码的监控。然后把相应的日志信息发送到后台。这里不展开。
服务器端埋点就相对容易，只需要拦截相应的请求，提取出关键信息，就能记录一些用户的日常行为，比如，通常该用户的登录、退出时间，哪些关键字被搜索更频繁等等。
这里我主要介绍一下几个地方：
- 请求的拦截；
- 将行为日志转储到kafka队列；
- 将日志提取出来存到数据库里。
关于请求的拦截，不同的java web框架的方式不太一样。这里贴一个spring mvc的示例。首先创建一个请求拦截类。
import java.util.Calendar;
import java.util.HashMap;
import java.util.Map;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.springframework.web.servlet.HandlerInterceptor;
import org.springframework.web.servlet.ModelAndView;
import com.alibaba.fastjson.JSON;
import com.cv.kdata.cont.RDDWebConst;
import com.cv.kdata.util.StringUtil;
import com.cv.kdata.util.TimeUtil;
import com.kdata.defined.model.TrackModel;
import com.kdata.track.KafkaProducerHelper;
public class TrackInterceptor implements HandlerInterceptor {
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)
throws Exception {
try {
String uri = request.getRequestURI();
// logout 操作的时候会清除session信息，所以在preHandle里面记录日志
if (!StringUtil.isNullOrEmpty(uri) && uri.contains("logout")) {
TrackModel model = new TrackModel();
model.setUri(request.getRequestURI());
if (request.getSession() != null) {
model.setSession((String) request.getSession().getAttribute(RDDWebConst.TOKEN));
model.setUserName((String) request.getSession().getAttribute(RDDWebConst.USERNAME));
}
Map<String,String> header = new HashMap<String,String>();
header.put("referer",request.getHeader("referer"));
header.put("origin",request.getHeader("origin"));
header.put("user-agent",request.getHeader("user-agent"));
model.setHeader(header);
model.setParas(request.getParameterMap());
model.setTime(TimeUtil.getCurrentTime());
model.setStatus(Integer.toString(response.getStatus()));
String time = Calendar.getInstance().getTime().toString();
KafkaProducerHelper.getInstance(RDDWebConst.KAFKA_HOST).send(RDDWebConst.TRACKTOPIC, time,
JSON.toJSONString(model));
}
} catch (Exception e) {
}
return true;
}
@Override
public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler,
ModelAndView modelAndView) throws Exception {
// System.out.println("postHandle");
}
@Override
public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex)
throws Exception {
try {
String uri = request.getRequestURI();
// logout的时候已经在prehandle里面记录了日志信息
if (!StringUtil.isNullOrEmpty(uri) && uri.contains("logout")) {
return;
}
TrackModel model = new TrackModel();
model.setUri(request.getRequestURI());
if (request.getSession() != null) {
model.setSession((String) request.getSession().getAttribute(RDDWebConst.TOKEN));
model.setUserName((String) request.getSession().getAttribute(RDDWebConst.USERNAME));
}
Map<String,String> header = new HashMap<String,String>();
header.put("referer",request.getHeader("referer"));
header.put("origin",request.getHeader("origin"));
header.put("user-agent",request.getHeader("user-agent"));
model.setHeader(header);
model.setParas(request.getParameterMap());
model.setTime(TimeUtil.getCurrentTime());
model.setStatus(Integer.toString(response.getStatus()));
String time = Calendar.getInstance().getTime().toString();
KafkaProducerHelper.getInstance(RDDWebConst.KAFKA_HOST).send(RDDWebConst.TRACKTOPIC, time,
JSON.toJSONString(model));
} catch (Exception e) {
}
}
}
这个类会把http请求的内容提取出来，然后发送到kafka队列。我采用kafka里String的序列化方式。为了最后实例化成对象更容易，可以先把对象转化成json, 然后传输json string。
现在最关键的就是让这个拦截器生效。我这边是在spring-mvc.xml里面配置。通过这个方式就可以把TrackInterceptor类注册到拦截器中。
<mvc:interceptor>
<mvc:mapping path="/**"/>
<bean class="com.test.interceptor.TrackInterceptor"></bean>
</mvc:interceptor>
现在已经完成了两个步骤，接下来就是从kafka队列里读取相应的内容。我这里读取出来直接解析到数据库里。
import java.util.Arrays;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.kd.configure.ConfigurationHelper;
import com.kd.consts.GeneralConst;
import com.kd.model.track.Generalinfo;
import com.kd.utils.KafkaConsumerHelper;
public class TrackInfo {
private static final Logger logger = LoggerFactory.getLogger(TrackInfo.class);
public static void main(String[] args) {
ConfigurationHelper.init();
new TrackInfo().syncTracking();
}
public void syncTracking() {
try {
logger.info("Start to sync tracking!");
String topic = GeneralConst.TRACKTOPIC;
KafkaConsumer<String, String> consumer = KafkaConsumerHelper.getInstance()
.getConsumer(GeneralConst.KAFKA_HOST, topic);
consumer.subscribe(Arrays.asList(topic));
int messagecounter = 0;
while (true) {
ConsumerRecords<String, String> records = consumer.poll(10000);
for (ConsumerRecord<String, String> record : records) {
Generalinfo model = JSON.parseObject(record.value(), Generalinfo.class);
if (model != null) {
model.save();
}
messagecounter++;
}
logger.info("received messages:" + messagecounter);
}
} catch (Exception e) {
logger.error("catch an error: " + e);
}
}
}
看到了吧，这里我用了另一个类Generalinfo 来实例化从kafka队列出来的string。这就是json的好处，可以增加代码的可移植性。
最后我们去看看数据库里的日志：
tracking
这个就是后台行为日志跟踪的大致过程。我将kafka的接口和日志提取就放到了github上。大家可以下载使用。只有请求拦截TrackInterceptor这个类需要自己写。代码是基于jfinal框架实现的，仅供参考。重要是解决思路上的问题。
