[分布式监控CAT] Server端源码解析——消息消费\报表处理\展示

前言

Server端

(Cat-consumer 用于实时分析从客户端提供的数据\Cat-home 作为用户给用户提供展示的控制端

，并且Cat-home做展示时，通过对Cat-Consumer的调用获取其他节点的数据，将所有数据汇总展示)

consumer、home以及路由中心都是部署在一起的，每个服务端节点都可以充当任何一个角色

Client端

(Cat-client 提供给业务以及中间层埋点的底层SDK)

相关文章：

[分布式监控CAT] Server端源码解析——初始化

[分布式监控CAT] Client端源码解析

[分布式监控CAT] Server端源码解析——消息消费\报表处理

上文说到了CAT-Server的启动初始化。

接着我们要分析一下CAT-Server如何接受各个客户端上报(TCP长连接)的消息，以及如何消费、解析、存储等等

先来看一下CAT整体的架构图：

消费、解析

com.dianping.cat.analysis.TcpSocketReceiver

在上一篇文章中说过了服务端的启动，在CAT-Server启动时会启动Netty的Nio 多线程Reactor模块来接收客户端的请求:

一个Accept线程池(Main Reactor Thread Pool )用来处理连接操作(通常还可以在这各Accept中加入权限验证、名单过滤逻辑)；

接着Accept连接成功的socket请求被转发到 专门处理IO操作的线程池（Sub Reactor Thread Pool ，实现异步)；在这里做了消息的解码处理；

再接着，解码处理后，将消息发送到每个报表解析器内置的内存队列中。消息将被异步分发给各个解析器单独处理(不存在数据竞争)。

消息的接受是在这个类TcpSocketReceiver.java完成的：

// 在CatHomeModule启动时被调用
public void init() {
try {
startServer(m_port);
} catch (Throwable e) {
m_logger.error(e.getMessage(), e);
}
}
/**
* 启动一个netty服务端
* @param port
* @throws InterruptedException
*/
public synchronized void startServer(int port) throws InterruptedException {
boolean linux = getOSMatches("Linux") || getOSMatches("LINUX");
int threads = 24;
ServerBootstrap bootstrap = new ServerBootstrap();
//linux走epoll的事件驱动模型
m_bossGroup = linux ? new EpollEventLoopGroup(threads) : new NioEventLoopGroup(threads);//用来做为接受请求的线程池 master线程
m_workerGroup = linux ? new EpollEventLoopGroup(threads) : new NioEventLoopGroup(threads);//用来做为处理请求的线程池 slave线程
bootstrap.group(m_bossGroup, m_workerGroup);
bootstrap.channel(linux ? EpollServerSocketChannel.class : NioServerSocketChannel.class);

bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {//channel初始化设置
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();

pipeline.addLast("decode", new MessageDecoder());//增加消息解码器
}
});
// 设置channel的参数
bootstrap.childOption(ChannelOption.SO_REUSEADDR, true);
bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

try {
m_future = bootstrap.bind(port).sync();//绑定监听端口,并同步等待启动完成
m_logger.info("start netty server!");
} catch (Exception e) {
m_logger.error("Started Netty Server Failed:" + port, e);
}
}

启动netty，对每个客户端上报的消息都会做解码处理，从字节流转换为消息树MessageTree tree，接着交给DefaultMessageHandler处理。

public class DefaultMessageHandler extends ContainerHolder implements MessageHandler, LogEnabled {

/*
* MessageConsumer按每个period(整小时一个period)组合了多个解析器，用来解析生产多个报表（如：Transaction、
* Event、Problem等等）。一个解析器对象-一个有界队列-一个整小时时间组合了一个PeriodTask，轮询的处理这个有界队列中的消息
*/
@Inject
private MessageConsumer m_consumer;

private Logger m_logger;

@Override
public void enableLogging(Logger logger) {
m_logger = logger;
}

@Override
public void handle(MessageTree tree) {
if (m_consumer == null) {
m_consumer = lookup(MessageConsumer.class);//从容器中加载MessageConsumer实例
}

try {
m_consumer.consume(tree);//消息消费
} catch (Throwable e) {
m_logger.error("Error when consuming message in " + m_consumer + "! tree: " + tree, e);
}
}
}

OMS设计是按照每小时去汇总数据，为什么要使用一个小时的粒度呢？

这个是一个trade-off，实时内存数据处理的复杂度与内存的开销方面的折中方案。

在这个小时结束后将生成的Transaction\Event\Problean报表存入Mysql、File（机器根目录侠）。然而为了实时性，当前小时的报表是保存在内存中的。

PeriodManager 用来管理 OMS单位小时内的各种类型的解析器，包括将上报的客户端数据派发给不同的解析器（这种派发可以理解为订阅\发布）。每个解析器，将收到的消息存入内置队列，并且用单独的线程去获取消息并处理。

接下来我们继续看代码：

com.dianping.cat.analysis.PeriodManager

public class PeriodManager implements Task {
public void init() {
long startTime = m_strategy.next(System.currentTimeMillis());//当前小时的起始时间

startPeriod(startTime);
}

@Override
public void run() {
// 1s检查一下当前小时的Period对象是否需要创建（一般都是新的小时需要创建一个Period代表当前小时）
while (m_active) {
try {
long now = System.currentTimeMillis();
//value>0表示当前小时的Period不存在，需要创建一个
//如果当前线小时的Period存在，那么Value==0
long value = m_strategy.next(now);
if (value > 0) {
startPeriod(value);
} else if (value < 0) {
//  //当这个小时结束后，会异步的调用endPeriod(..)，将过期的Period对象移除，help GC
Threads.forGroup("cat").start(new EndTaskThread(-value));
}
} catch (Throwable e) {
Cat.logError(e);
}

try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
break;
}
}
}
//当这个小时结束后，会异步的调用这个方法，将过期的Period对象移除，help GC
private void endPeriod(long startTime) {
int len = m_periods.size();

for (int i = 0; i < len; i++) {
Period period = m_periods.get(i);

if (period.isIn(startTime)) {
period.finish();
m_periods.remove(i);
break;
}
}
}

......
}

消息消费是由MessageConsumer的实现类RealtimeConsumer处理：

com…RealtimeConsumer.consume(MessageTree tree)

@Override
public void consume(MessageTree tree) {
String domain = tree.getDomain();
String ip = tree.getIpAddress();

if (!m_blackListManager.isBlack(domain, ip)) {// 全局黑名单 按domain-ip
long timestamp = tree.getMessage().getTimestamp();
//PeriodManager用来管理、启动periodTask,可以理解为每小时的解析器。
Period period = m_periodManager.findPeriod(timestamp);//根据消息产生的时间，查找这个小时所属的对应Period

if (period != null) {
period.distribute(tree);//将解码后的tree消息依次分发给所有类型解析器
} else {
m_serverStateManager.addNetworkTimeError(1);
}
} else {
m_black++;

if (m_black % CatConstants.SUCCESS_COUNT == 0) {
Cat.logEvent("Discard", domain);
}
}
}

分发消息给各个解析器（类似向订阅者发布消息）

void com.dianping.cat.analysis.Period.distribute(MessageTree tree)

/**
* 将解码后的tree消息依次分发给所有类型解析器
* @param tree
*/
public void distribute(MessageTree tree) {
m_serverStateManager.addMessageTotal(tree.getDomain(), 1);// 根据domain,统计消息量
boolean success = true;
String domain = tree.getDomain();

for (Entry<String, List<PeriodTask>> entry : m_tasks.entrySet()) {
List<PeriodTask> tasks = entry.getValue();//某种类型报表的解析器
int length = tasks.size();
int index = 0;
boolean manyTasks = length > 1;

if (manyTasks) {
index = Math.abs(domain.hashCode()) % length;//hashCode的绝对值 % 长度 =0~length-1之间的任一个数
}
PeriodTask task = tasks.get(index);
boolean enqueue = task.enqueue(tree);//注意:这里会把同一个消息依依放入各个报表解析中的队列中

if (enqueue == false) {
if (manyTasks) {
task = tasks.get((index + 1) % length);
enqueue = task.enqueue(tree);//放入队列，异步消费

if (enqueue == false) {
success = false;
}
} else {
success = false;
}
}
}

if (!success) {
m_serverStateManager.addMessageTotalLoss(tree.getDomain(), 1);
}
}

PeriodTask

每个periodTask对应一个线程，m_analyzer对应解析器处理m_queue中的消息

public class PeriodTask implements Task, LogEnabled {
@Override
public void run() {//每个periodTask对应一个线程，m_analyzer对应解析器处理m_queue中的消息
try {
m_analyzer.analyze(m_queue);
} catch (Exception e) {
Cat.logError(e);
}
}

AbstractMessageAnalyzer

@Override
public void analyze(MessageQueue queue) {// 解析器在当前小时内自旋，不停从队列中拿取消息，然后处理
while (!isTimeout() && isActive()) {// timeOut:当前时间>小时的开始时间+一小时+三分钟；
// isActive默认为true，调用shutdown后为false
MessageTree tree = queue.poll();// 非阻塞式获取消息

if (tree != null) {
try {
process(tree);// 解析器实现类 override
} catch (Throwable e) {
m_errors++;

if (m_errors == 1 || m_errors % 10000 == 0) {
Cat.logError(e);
}
}
}
}
// 如果当前解析器以及超时，那么处理完对应队列内的消息后返回。
while (true) {
MessageTree tree = queue.poll();

if (tree != null) {
try {
process(tree);
} catch (Throwable e) {
m_errors++;

if (m_errors == 1 || m_errors % 10000 == 0) {
Cat.logError(e);
}
}
} else {
break;
}
}
}

所以我们可以看到：

消息发送到服务端，服务端解码为 MessageTree准备消费。期间存在一个demon线程，1s检查一下当前小时的Period对象是否需要创建（一般都是新的小时需要创建一个Period代表当前小时）。

如果当前小时的Period存在，那么我们的MessageTree会被分发给各个PeriodTask，这里其实就是把消息发送到每个PeriodTask中的内存队列里，然后每个Task异步去消费。就是通过使用Queue实现了解耦与延迟异步消费。

每个PeriodTask持有MessageAnalyzer analyzer(Transaction\Event\Problean…每种报表都对应一个解析器的实现类)、MessageQueue queue对象，PeriodTask会不停地解析被分发进来的MessageTree，形成这个解析器所代表的报表。

当前时间进入下个小时，会创建一个新的当前小时的Period，并且异步的remove之前的Period。

注意，这里有个比较坑的地方是，作者没有使用线程池，每小时各个解析器的线程并没有池化，而是直接销毁后再次创建！

展示

对于实时报表，直接通过HTTP请求分发到相应消费机，待结果返回后聚合展示（对分区数据做聚合）；历史报表则直接取数据库并展示。

存储

存储主要分成两类：一个是 报表（Transaction、Event、Problem….），一个是logview,也是就是原始的MessageTree。

所有原始消息会先存储在本地文件系统，然后上传到HDFS中保存；而对于报表，因其远比原始日志小，则以K/V的方式保存在MySQL中。

报表存储：在每个小时结束后，将内存中的各个XML报表 保存到Mysql、File（\data\appdatas\cat\bucket\report…）中。



logView的保存有后台线程（默认20个，Daemon Thread [cat-Message-Gzip-n]）轮询处理：会间隔一段时间后从消息队列中拿取MessageTree，并进行编码压缩，保存到\data\appdatas\cat\bucket\dump\年月\日\domain-ip1-ip2-ipn目录下。

com.dianping.cat.consumer.dump.LocalMessageBucketManager.MessageGzip.run()

@Override
public void run() {
try {
while (true) {
MessageItem item = m_messageQueue.poll(5, TimeUnit.MILLISECONDS);

if (item != null) {
m_count++;
if (m_count % (10000) == 0) {
gzipMessageWithMonitor(item);//数量达到10000的整数倍，通过上报埋点记录监控一下
} else {
gzipMessage(item);
}
}
}
} catch (InterruptedException e) {
// ignore it
}
}

private void gzipMessage(MessageItem item) {
try {
MessageId id = item.getMessageId();
String name = id.getDomain() + '-' + id.getIpAddress() + '-' + m_localIp;
String path = m_pathBuilder.getLogviewPath(new Date(id.getTimestamp()), name);
LocalMessageBucket bucket = m_buckets.get(path);

if (bucket == null) {
synchronized (m_buckets) {
bucket = m_buckets.get(path);
if (bucket == null) {
bucket = (LocalMessageBucket) lookup(MessageBucket.class, LocalMessageBucket.ID);
bucket.setBaseDir(m_baseDir);
bucket.initialize(path);

m_buckets.put(path, bucket);
}
}
}

DefaultMessageTree tree = (DefaultMessageTree) item.getTree();
ByteBuf buf = tree.getBuffer();
MessageBlock block = bucket.storeMessage(buf, id);

if (block != null) {
if (!m_messageBlocks.offer(block)) {
m_serverStateManager.addBlockLoss(1);
Cat.logEvent("DumpError", tree.getDomain());
}
}
} catch (Throwable e) {
Cat.logError(e);
}
}

public MessageBlock storeMessage(final ByteBuf buf, final MessageId id) throws IOException {
synchronized (this) {
int size = buf.readableBytes();

m_dirty.set(true);
m_lastAccessTime = System.currentTimeMillis();
m_blockSize += size;
m_block.addIndex(id.getIndex(), size);
buf.getBytes(0, m_out, size); // write buffer and compress it

if (m_blockSize >= MAX_BLOCK_SIZE) {
return flushBlock();
} else {
return null;
}
}
}

logView的文件存储设计

接下来，会介绍CAT中出现的一些经典的设计、算法。