一种基于kafka+storm实现的日志记录方法

背景：

 

在我们系统中，每逢大促经常会遇到活动页面莫名被篡改的情况，有些操作是人为故意修改，比如大促时把某些坑位的商品全部换为某个商家的商品，这里就存在恶意修改页面的问题。每次遇到这种问题，内控合规部 需要找我们提取操作日志，追溯所有篡改记录，查找元凶。

 

由于活动页面装修可以通过授权给多人装修，我们mysql表里的操作日志只有最后一次修改的状态记录，没有操作流水日志。无法从技术上排除每个被授权人的嫌疑。

 

当然你可以会觉得解决这个问题很简单，可以首先通过传统的打印日志的方法。但这个种方式有如下缺陷：

1、日志散落在各台应用服务器，要收集齐所有相关日志非常麻烦。

2、由于服务器硬盘资源有限，日志超过一定量或多少天就会被删除，如超过7天就会被删除。

3、我们公司不允许在日志中打印用户的账户等敏感信息。这点是最致命的。

 

也许你还会想到直接把日志写入的nosql数据库的方式，比如 Mongodb、ES、hbase等nosql数据库。但直接把日志写入到数据库的方式存在很大的性能问题，每打印一条日志都需要消耗几十到几百毫秒不等。严重影响正常业务的性能。

 

采用kafka+storm异步记录日志方式

 

借鉴点击流日志上报流程，只是上报客户端从用户的浏览器变成了我们的系统服务器，上报的内容从浏览数据变成了用户操作日志。使用到的技术：java+kafka+storm+hbase。具体流程如下：

1、前期准备工作：预先为每个需要接入该“日志上报工具”的业务系统分配一个系统id，如”sys_A”。在我们的实现中，有一个“业务接入”注册功能：

 

2、把用java语言实现的向kafka集群指定的“日志topic”发送消息的功能，封装成一个通用的jar包，在每个需要接入该“日志上报工具”的业务系统中引入该jar包。

 

3、在需要打印用户操作日志的地方，收集系统id、用户信息（账号，ip等）、查询key、用户操作、日志描述、操作时间等信息 组装成一个map。调用通用jar包中的日志打印方法SelfLog.log(map)。

 

4、在通用jar包SelfLog.log（Map map）方法中，会检查日志参数是否正确（比如，是否包含关键的 “系统id”和“查询key”，因为查询日志时需要根据这两个参数做定位）。检查通过后，把map转换成json字符串发送到指定kafka集群的指定topic中。

向kafka发送消息是个很快的过程，性能消耗几乎可以忽略不计，对正常业务几乎没有影响。同时利用kafka的高吞吐量，每天支持亿级的日志量没有问题。

 

5、通过storm实时消费kafka的指定“日志topic”，对日志进行解析，并根据“系统id”进行分组，每隔指定时间（比如1分钟）把收集到的日志信息，批量写入hbase的指定表中。这里每隔指定时间向hbase批量写入，主要是为了减少与hbase的交互次数，提升性能。这里指定的间隔时间，就是日志查询的延迟时间，可以做到近乎实时的日志查询。

日志被写入hbase日志表以后，就可以通过编写一个查询hbase的web页面进行日志查询。通过录入相关查询key，一次就可以提取出所有相关的日志信息：

最终效果

 

1、实现高效的流水日志记录（几乎没有性能消耗）。方便问题追溯。

2、可以打印敏感信息，这些信息最终是存在到hbase表，而不是写到日志文件。

 

3、相比传统的日志打印方式，日志存储周期更长，可以做到永久存在。日志不再散落在各个应用服务器，查询日志更方便直观。

 

为了提高日志的查询效率，我们把“查询key”做为hbase日志表rowkey的一部分，所有每次需要打印日志时，需要认真定义这个“查询key”，比如操作某个活动，可以把活动id作为“查询key”。

 

最后，通过创建更多的“topic”，在storm消费的地方再进行更细分处理，再加上一些权限验证，完全可以把这个工具平台化，提供一个类似“统一日志管理”的日志平台。

评论

11 楼 moon_walker 2018-01-19  

果粒儿我是 写道

“切面方法里只能接收到方法的入参，也就是你这里的UserDTO，可以把入参的内容统一写入消息队列。但切面里取到的入参是Object类型，需要一次类型转换。” 如果只是 切面get入参的数据，可以不用放到消息队列or redis, 再定义一个注解修饰UserDTO你想要什么属性就好了。 然后切面get到这个属性就ok

没有明白你的这段的意思。感觉你就是对delete的场景不好处理，为了简单起见还是建议在方法名上做些规范。比如把你的delete(int userId)方法改为：
deleteUserById(int userId)

在切面方法里可以取到 方法名和入参，可以拼装你需要的日志 delete User表中id为xxx的记录。 在把当前用户信息放入ThreadLocal，在切面方法里取到当前用户，就可以实现：
who delete User表中id为xxx
不知道你明白我说的意思没，感觉可以解决你的问题。

10 楼 果粒儿我是 2018-01-19  

“切面方法里只能接收到方法的入参，也就是你这里的UserDTO，可以把入参的内容统一写入消息队列。但切面里取到的入参是Object类型，需要一次类型转换。” 如果只是 切面get入参的数据，可以不用放到消息队列or redis, 再定义一个注解修饰UserDTO你想要什么属性就好了。 然后切面get到这个属性就ok

9 楼 果粒儿我是 2018-01-19  

moon_walker 写道

创建基类也有侵入性，感觉可以把方法名规范化。比如selectXXX，delectXXX，在切面中可以获取到方法名是delete开头的，就可以认为是删除操作，新增和查询类似。 这也就是所谓的约定大于配置的编程方式吧。

创建基类的方式的确有侵入性，而且违背了aop初衷貌似..
我想要做的是为微服务多个服务提供一个包，专门写日志，使用者只需要在自己的应用里加入注解就够了。但是这种只适合“通用”日志，不适合我的场景 。

通用日志是，例如异常日志，错误日志等等。就是所有日志的格式固定。我的日志场景是想提取出来用户都操作了什么，然后展示给用户看。举例子来说就是删除。前端传入删除用户id = 5.
  @Auditable(actionType = ActionType.DELETE, subject = Subject.User)
    @RequestMapping("/delete")
    @ResponseBody
    public void
    delete(int userId){
       String userNameDeleted = userService.getUserById(userId);// 需要记录日志 “DELETE USER 张三“，
        userInfoService.delete(userId);
    }
需要记录日志 “DELETE USER 张三“，  但是这个张三切面是无法捕获到的，因为它不是注解标记的方法的入参或者出参，他更需要传递给切面。所以貌似切面无法满足这种动态一点的日志需求。

8 楼 moon_walker 2018-01-19  

创建基类也有侵入性，感觉可以把方法名规范化。比如selectXXX，delectXXX，在切面中可以获取到方法名是delete开头的，就可以认为是删除操作，新增和查询类似。 这也就是所谓的约定大于配置的编程方式吧。

7 楼 moon_walker 2018-01-19  

果粒儿我是 写道

最近也有遇到类似场景，这个场景实际上是一个关于如何写应用系统的审计日志的问题。使用MQ+存储工具貌似是目前发现比较通用的方法哈。

关于如何收集日志我有一点疑问想请教哈，我看文中是通过在应用业务程序中加入SelfLog.log(map)，这种写法其实对业务代码侵入性很大，因为写业务代码的人员只需要关注写业务就好了。 我想到一个办法是使用切面+自定义注解实现这个写审计日志功能。例如当我在系统新增一个用户，我可以实现如下代码：

    @Auditable(actionType = ActionType.ADD, subject = Subject.User)
    @RequestMapping("/signup")
    @ResponseBody
    public void
    signUp(UserDTO userDTO){
        UserInfo userInfo = new UserInfo();
        userInfo.setName(userDTO.getName());
        userInfo.setPassword("11");
        userInfo.setUid(1);
        userInfo.setState(1);
        userInfoService.insert(userInfo);

    }
通过在切面中可以获取操作类型是新增，操作主体是用户，等等信息。但是有一个问题，如果我想删除用户，前端传递删除的用户id=5, 因为切面是一种统一操作，他是没法获取用户名称，因此也没法实现写日志“删除用户张三”。

请问如何能实现向切面传递动态信息呢？

你这个想法不错啊，我知道的切面方法里只能接收到方法的入参，也就是你这里的UserDTO，可以把入参的内容统一写入消息队列。但切面里取到的入参是Object类型，需要一次类型转换。如果入参的类型很多 这时切面就不通用了，最好在定义一个基类，让入参类都继承这个基类，按照这个方法理论上可行。我也想按照你的想法改进下，谢谢

6 楼 果粒儿我是 2018-01-18  

最近也有遇到类似场景，这个场景实际上是一个关于如何写应用系统的审计日志的问题。使用MQ+存储工具貌似是目前发现比较通用的方法哈。

关于如何收集日志我有一点疑问想请教哈，我看文中是通过在应用业务程序中加入SelfLog.log(map)，这种写法其实对业务代码侵入性很大，因为写业务代码的人员只需要关注写业务就好了。 我想到一个办法是使用切面+自定义注解实现这个写审计日志功能。例如当我在系统新增一个用户，我可以实现如下代码：

    @Auditable(actionType = ActionType.ADD, subject = Subject.User)
    @RequestMapping("/signup")
    @ResponseBody
    public void
    signUp(UserDTO userDTO){
        UserInfo userInfo = new UserInfo();
        userInfo.setName(userDTO.getName());
        userInfo.setPassword("11");
        userInfo.setUid(1);
        userInfo.setState(1);
        userInfoService.insert(userInfo);

    }
通过在切面中可以获取操作类型是新增，操作主体是用户，等等信息。但是有一个问题，如果我想删除用户，前端传递删除的用户id=5, 因为切面是一种统一操作，他是没法获取用户名称，因此也没法实现写日志“删除用户张三”。

请问如何能实现向切面传递动态信息呢？

5 楼 moon_walker 2017-07-21  

QuarterLifeForJava 写道

可否讲解下Java如和生产消息到Kafka，两边的配置及核心代码，以前storm和kafa的配置和核心代码，还有HBase，如果能再将就下配置、编写代码中的坑那就更好了

没问题，周末先把java如何生产消息到kafka整理出来

4 楼 QuarterLifeForJava 2017-07-21  

可否讲解下Java如和生产消息到Kafka，两边的配置及核心代码，以前storm和kafa的配置和核心代码，还有HBase，如果能再将就下配置、编写代码中的坑那就更好了

3 楼 xierui 2017-07-21  

很不错,学习了

2 楼 moon_walker 2017-07-20  

sg6303 写道

有源码可以提供学习吗？？

kafka和storm都是用公司的资源，代码含有公司业务，需要剥离才能公开。其实实现起来也不难，主要就是编写一个java往kafka写消息的生成者，再使用storm消费这些消息。storm的主要职责就是对消息归类，以及批量存储到hbase。
如果大家需求强烈，我整理下，可以公开部分代码。

1 楼 sg6303 2017-07-20  

有源码可以提供学习吗？？