Web系统后台行为记录

---

关于后台行为记录，可理解为对用户请求的监控日志。也可理解为埋点的一种。通常埋点可简单分为前端埋点和后端埋点。

前端埋点需要对APP或Web的一些控件进行有代码或无代码的监控。然后把相应的日志信息发送到后台。这里不展开。

服务器端埋点就相对容易，只需要拦截相应的请求，提取出关键信息，就能记录一些用户的日常行为，比如，通常该用户的登录、退出时间，哪些关键字被搜索更频繁等等。
这里我主要介绍一下几个地方：

1. 请求的拦截；
2. 将行为日志转储到kafka队列；
3. 将日志提取出来存到数据库里。

关于请求的拦截，不同的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框架实现的，仅供参考。重要是解决思路上的问题。