背景介绍

随着业务服务（Server App）逐渐增加，使得问题排查非常困难，很多时候需要关联查询多个服务的日志，而且统计分析十分不便。因此，急需设计一个集中式海量日志实时处理系统。需要满足功能需求（实时看日志、统计历史日志、实时行为分析、用户轨迹跟踪等）、性能需求（具有高吞吐能力、高扩展性、高容错性）等。

组件介绍

Kafka 是一种高吞吐量的分布式发布订阅消息系统，它适合处理海量日志发布订阅，提供消息磁盘持久化、支持物理分片存储、多组消费等特性。

Elasticsearch 是一个开源实时分布式搜索引擎，具备如下特征：零配置，索引自动分片，索引副本机制，restful风格接口，多数据源，自动搜索负载等。

Flume 是Apache基金会的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统，它支持在日志系统中定制各类数据发送方，用于收集数据；同时提供对数据进行简单处理，并写到各种数据接受方（可定制）。

实现思路

1. 开发业务日志SDK（下文为描述方便，称之为 BizLogSDK），嵌于各业务App；
2. 从业务服务端收集日志并集中输出到Kafka；
3. 根据不同需求（查询、统计），由Flume对数据预处理并分发；
4. Flume 的下游组件对日志内容进行消费；

架构设计

业务日志服务架构.png

目前日志消费方式有两种：

• Elasticsearch 做索引，用于查询、统计；
• 基于Storm流式计算实现（待完成）。

引入Kafka的目的：

• 线上业务集群规模较大，日志产生量巨大，如果直接同步日志对下游服务负荷较重，容易因为故障导致日志阻塞延迟和丢失，所以引入了kafka ；
• 消息可以持久化，并且可以进行日志回溯。有了消息队列，上游服务和下游服务完全解藕，网络传输会更稳定、更高效、更均衡，避免级联效应。

架构说明

1. 由于业务日志量极大，为减轻业务服务的压力，故将业务日志首先输出到 Kafka 集群；
2. Flume 做分发和预处理。从Kafka中拉取待处理的业务日志，先在本地保留一份，然后做预处理和分发；
3. Elasticsearch 做日志索引。对业务日志按 Prefix-bizType-YYYY.MM.dd 的格式创建索引；
4. Kibana 做查询界面与简单的统计报表。供开发、运维、运营人员使用；
5. Zookeeper 用于维护Kafka集群配置。Flume作为Kafka的消费者，需要配置Zookeeper的相关信息；
6. Kibana 的报表展示能力有限，可以在Elasticsearch 下游对接 Grafana或其他工具（架构图中未做描述），实现更炫酷的报表；
7. 可以根据业务扩展需求，增加对应的 Flume 及处理服务，以实现业务横向扩展；
8. 目前还没有对业务日志做大数据分析，因此架构中只做了节点描述。

技术方案

1. Flume 1.6增加了 KafkaSource，之前版本需要自己实现（自定义 Source 实现示例）；
2. Flume 做预处理和分发，需要自定义Sink（自定义Sink实现示例）；
3. BizLogSDK 的配置中可以添加对Kafka Producer 的配置（Producer Configs），以优化性能；
4. Elasticsearch 官网的 Java Client 比较重，连接数太多，建议按照Elasticsearch Reference 自己开发一个基于HTTP 协议 的 Client（实现CRUD），方便业务日志按照 biztype-date 格式进行索引。注意：HTTP 连接应该复用，（可以采用HttpClient 的连接池管理方式），避免连接数过多；
5. Flume Sink 中拿到业务日志后，应该放到线程池里处理，避免 Flume卡死；
6. Elasticsearch 默认会将 string 类型字段设为 _ analyzed _，会造成CPU过高。可以通过 Elasticsearch 提供的 Index Templates 方式，在index 创建后，应用 template 到匹配的index，将相关 _ string _ 型字段设为 _ not_analyzed _。

BizLogSDK ：

业务App通过调用 BizLogSDK，将业务日志输出到Kafka集群。BizLogSDK 需要实现设置公用属性、扩展属性，日志发送等功能。可参考 Log4j 的源码来实现。

1. 业务日志公用属性：

• _ bizType _ ：业务类型
• _ bizAction _：业务操作
• _ serverIp _ ：服务器IP
• _ requestTime _ ：请求时间

2. 基本配置：

# 队列名称
topicName = bizlog-server
# 是否同步发送消息（异步速度更快）
send.sync = false
# kafka 消息队列服务器
bootstrap.servers = 127.0.0.1:9091

实时查询

查询界面

实时统计

统计界面 1

统计界面 2

相关监控

• KafkaOffsetMonitor

Kafka队列监控

• Elasticsearch Monitor

  
  
  Elasticsearch 监控

后续改进

• BizLogSDK 中需要加入 Log Level；
• 业务日志需要一个统一的界面来管理（设置level、关闭、删除、定期清理等）；