深入详解美团点评CAT跨语言服务监控（七）消息分析器与报表(二)

CrossAnalyzer-调用链分析

在分布式环境中，应用是运行在独立的进程中的，有可能是不同的机器，或者不同的服务器进程。那么他们如果想要彼此联系在一起，形成一个调用链，在Cat中，CrossAnalyzer会统计不同服务之间调用的情况，包括服务的访问量，错误量，响应时间，QPS等，这里的服务主要指的是 RPC 服务，在微服务监控中，这是核心。

在讲 CrossAnalyzer 的处理逻辑之前，我们先看下客户端的埋点的一个模拟情况。

一般情况下不同服务会通过几个ID进行串联。这种串联的模式，基本上都是一样的。在Cat中，我们需要3个ID：

RootId，用于标识唯一的一个调用链
ParentId，父Id是谁？谁在调用我
ChildId，我在调用谁?

那么我们如何传递这些ID？Cat为我们提供了一个内部接口 Cat.Context，但是我们需要自己实现Context，在下面代码中我们首先在before函数中实现了Context 上下文，然后在rpcClient中开启消息事务，并调用 Cat.logRemoteCallClient(context) 去填充Context的这3个MessageID。当然，该函数还记录了一个RemoteCall类型的Event消息。

随后我们用rpcService函数中开启新线程模拟远程RPC服务，并将context上传到 RPC 服务器，在真实环境中，Context是需要跨进程网络传输，因此需要实现序列化接口。

在rpcService中，我们会调用 Cat.logRemoteCallServer(context) 将从rpcClient传过来的Context设置到自己的 Transaction 当中。

随着业务处理逻辑的结束， rpcServer 和 rpcClient 都会分别将自己的消息树上传到CAT服务器分析。

需要注意的是，Service的client和app需要和Call的server以及app对应上，要不然图表是分析不出东西的！

@RunWith(JUnit4.class)
public class AppSimulator extends CatTestCase {
    public Map<String, String> maps = new HashMap<String, String>();

    public Cat.Context context;

    @Before
    public void before() {
        context = new Cat.Context() {
            @Override
            public void addProperty(String key, String value) { maps.put(key, value); }

            @Override
            public String getProperty(String key) { return maps.get(key); }
        };
    }

    @Test
    public void simulateHierarchyTransaction() throws Exception {
            ...
            //RPC调用开始
            rpcClient();
            rpcClient2();
            ...
    }

    protected void rpcClient() {
        //客户端埋点，Domain为RpcClient，调用服务端提供的Echo服务
        Transaction parent = Cat.newTransaction("Call", "CallServiceEcho");
        Cat.getManager().getThreadLocalMessageTree().setDomain("RpcClient");

        Cat.logEvent("Call.server","localhost");
        Cat.logEvent("Call.app","RpcService");
        Cat.logEvent("Call.port","8888");
        Cat.logRemoteCallClient(context, "RpcClient");

        //开启新线程模拟远程RPC服务，将context上传到 RPC 服务器
        rpcService(context);

        parent.complete();
    }

    protected void rpcClient2() {
        ...
        //模拟另外一个RpcClient调用Echo服务
        rpcService(context, "RpcClient2");
        ...
    }

    protected void rpcService(final Cat.Context context, final String clientDomain) {
        Thread thread = new Thread() {
            @Override
            public void run() {
                //服务器埋点，Domain为 RpcService 提供Echo服务
                Transaction child = Cat.newTransaction("Service", "Echo");
                Cat.getManager().getThreadLocalMessageTree().setDomain("RpcService");

                Cat.logEvent("Service.client", localhost); //填客户端地址
                Cat.logEvent("Service.app", clientDomain);
                Cat.logRemoteCallServer(context);

                //to do your business

                child.complete();
            }
        };

        thread.start();

        try {
            thread.join();
        } catch (InterruptedException e) {
        }
    }
}

接下来我们看看CAT服务器端CrossAnalyzer的逻辑。

我们依然会为每个周期时间内的每个Domain创建一张报表(CrossReport)，然后不同的IP会分配不同的Local对象统计，每个IP又可能会接收来自不同Remote端的调用。

由于这里一个完整的调用链会涉及多个端的多个消息树，我们首先会根据Transaction的类型来判断是RpcService还是RpcClient，如果Type等于PigeonService或Service则该消息来自RpcService，如果Type等于 PigeonCall或Call则来自RpcClient。

先来看看RpcService端消息树的上报处理逻辑，CAT会调用 parsePigeonServerTransaction 函数去填充 CrossInfo 信息，CrossInfo包含的具体内容如下：

localAddress ： RpcService的IP地址

remoteAddress ： 服务调用者(RpcClient)的IP地址，由type="Service.client" 的Event子消息提供，注意，在处理RpcClient的上报时，我们会根据上报信息中的remoteAddress再次统计该RpcService数据，大家可能会疑惑这里是不是重复统计，事实上他们所处的视角是不一样的，前者是站在服务提供者的视角来统计我完成这次服务所耗费的时间、资源等，而后者则是站在RpcClient视角去统计自己从发出请求到得到结果所需的时长、资源等等，比如这中间就包含网络IO的消耗，这些在后续的报表中会有体现。

app：客户端的Domain， 由type="Service.app"的Event子消息提供。

remoteRole：固定为 Pigeon.Client ， 表示远端角色为 Rpc 客户端。

detailType： 固定为 PigeonService ， 表示自己角色为 Rpc 服务端。

最后，我们将用CrossInfo信息来更新报表(CrossReport)，我们首先根据 localAddress 即 RpcService的 IP 找到或创建 Local对象，然后根据 remoteAddress+remoteRole 找到或创建 Remote 对象，然后统计服务的访问量，错误量，处理时间，QPS。

RpcService提供不只一个服务，不同的服务我们按名字分别统计在不同的Name对象里，比如上面案例，RpcService提供的是Echo服务。

我们再来看看RpcClient端上报处理逻辑，CAT调用parsePigeonClientTransaction函数填充CrossInfo信息，具体如下：

localAddress ： RpcClient的IP地址

remoteAddress ：服务提供者(RpcService)的地址，由 type="Call.server" 的Event子消息提供。

app：服务提供者的Domain，由type="Call.app" 的Event子消息提供，在统计完RpcClient端数据之后，会通过该属性获取服务提供者的CrossInfo。从RpcClient的视角再次统计RpcService的数据。

port：客户端端口，由 type="Call.port" 的Event子消息提供。

remoteRole：固定为 Pigeon.Server， 表示远端角色为服务提供者。

detailType： 固定为 PigeonCall ， 表示自己角色为服务调用者。

然后，我们将用CrossInfo信息来更新报表(CrossReport)，也是根据 localAddress 找到Local对象，然后根据 remoteAddress+remoteRole 找到 Remote 对象，进行统计。

接着，我们通过convertCrossInfo函数利用RpcClient的CrossInfo信息去生成服务提供者的CrossInfo信息，这里实际上是为了从RpcClient的视角去统计服务提供者的报表！

public class CrossAnalyzer extends AbstractMessageAnalyzer<CrossReport> implements LogEnabled {
    private void processTransaction(CrossReport report, MessageTree tree, Transaction t) {
        CrossInfo crossInfo = parseCorssTransaction(t, tree);

        if (crossInfo != null && crossInfo.validate()) {
            updateCrossReport(report, t, crossInfo);

            String targetDomain = crossInfo.getApp();

            if (m_serverConfigManager.isRpcClient(t.getType()) && !DEFAULT.equals(targetDomain)) {
                CrossInfo serverCrossInfo = convertCrossInfo(tree.getDomain(), crossInfo);

                if (serverCrossInfo != null) {
                    CrossReport serverReport = m_reportManager.getHourlyReport(getStartTime(), targetDomain, true);

                    updateCrossReport(serverReport, t, serverCrossInfo);
                }
            } else {
                m_errorAppName++;
            }
        }
        ...
    }
}

这里的 serverCrossInfo 被填充了什么数据：

localAddress ： RpcClient 的 remoteAddress。

remoteAddress ：RpcClient 的 localAddress + clientPort

app：RpcClient 的 Domain。

remoteRole：固定为 Pigeon.Caller， 表示远端角色为服务调用者。

detailType： 固定为 PigeonCall

最后还是用CrossInfo信息来更新报表(CrossReport)。

最后我们看看我们生成了哪些报表数据，3个报表数据，分别是服务调用方 RpcClient和 RpcClient2，以及服务提供方RpcService。

接下来我们看看服务提供方的remotes数据信息，一共3条数据，第1条记录是站在RpcService角度统计服务器完成这2次服务所耗费的时间、资源等，后面2条记录则是站在RpcClient视角去统计自己从发出请求到得到结果所需的时长、资源等等。

第1条记录 duration 为 0.154ms， 第2,3条记录 duration 分别为 1072.62ms、1506.38ms， 两者巨大的时间差一般就是网络 IO 所需的时间，事实上大多数的服务时间的消耗都是在各种IO上。这类服务统称为IO密集型。

StorageAnalyzer  --数据库/缓存分析

StorageAnalyzer主要分析一段时间内数据库、Cache访问情况：各种操作访问次数、响应时间、错误次数、长时间访问量等等，当客户端消息过来，StorageAnalyzer首先会分析事务属于数据库操作还是缓存操作，然后进行不同的处理，消息类型如果是SQL则是数据库操作，如果以Cache.memcached开头则认为是缓存操作。

我们首先看看数据库操作的分析过程，下面源码是客户端的案例，这是一个获取cat库config表全部数据的sql查询，我们将数据库操作所有信息都放在一个type="SQL" 的子事务消息中。

@RunWith(JUnit4.class)
public class AppSimulator extends CatTestCase {
    @Test
    public void simulateHierarchyTransaction() throws Exception {
        ...
        Transaction sqlT = cat.newTransaction("SQL", "Select");

        //do your SQL query

        cat.logEvent("SQL.Database", "jdbc:mysql://192.168.20.67:3306/cat");
        cat.logEvent("SQL.Method", "select");
        cat.logEvent("SQL.Statement", "SELECT", SUCCESS, "select * from cat.config");
        sqlT.complete();
        ...
    }
}

上面消息上报到服务端之后，分析器将SQL类型子事务取出，调用processSQLTransaction去处理，将结果写入报表StorageReport

processSQLTransaction 首先通过DatabaseParser提取数据库的IP和数据库名称，该信息由type="SQL.Database"的Event子消息提供，该Event消息上报的是数据库连接的URL。

接着我们会获取数据库操作名，type="SQL.Method" 的Event子消息提供，数据库操作分4类，分别是select, update, delete, insert，如果不上报，分析器默认客户端在做select查询。

最后我们会为周期内的每个数据库创建一个Storage报表。并将提取信息放入StorageUpdateParam对象，然后将对象交给StorageReportUpdater来更新Storage报表。

public class StorageAnalyzer extends AbstractMessageAnalyzer<StorageReport> implements LogEnabled {
    @Inject
    private DatabaseParser m_databaseParser;

    @Inject
    private StorageReportUpdater m_updater;

    private void processSQLTransaction(MessageTree tree, Transaction t) {
        String databaseName = null;
        String method = "select";
        String ip = null;
        String domain = tree.getDomain();
        List<Message> messages = t.getChildren();

        for (Message message : messages) {
            if (message instanceof Event) {
                String type = message.getType();

                if (type.equals("SQL.Method")) {
                    method = message.getName().toLowerCase();
                }
                if (type.equals("SQL.Database")) {
                    Database database = m_databaseParser.queryDatabaseName(message.getName());

                    if (database != null) {
                        ip = database.getIp();
                        databaseName = database.getName();
                    }
                }
            }
        }
        if (databaseName != null && ip != null) {
            String id = querySQLId(databaseName);
            StorageReport report = m_reportManager.getHourlyReport(getStartTime(), id, true);
            StorageUpdateParam param = new StorageUpdateParam();

            param.setId(id).setDomain(domain).setIp(ip).setMethod(method).setTransaction(t)
                  .setThreshold(LONG_SQL_THRESHOLD);// .setSqlName(sqlName).setSqlStatement(sqlStatement);
            m_updater.updateStorageReport(report, param);
        }
    }
}

数据库与缓存的报表更新逻辑相同，不同ip地址的数据库/缓存的统计信息在不同Machine里面，同时也可能会有不同的Domain访问同一个数据库/缓存，每个Domain的访问都会被单独统计，每个Domain对数据库/缓存不同的操作会统计在不同Operation里，除了当前小时周期的统计汇总外，我们还会用Segment记录每分钟的汇总数据。访问时间超过1秒的数据库操作(缓存是50ms) 会被认为是长时间访问记录。

缓存操作

接下来我们看下缓存的案例，获取memcached中key="uid_1234567"的值，Storage分析器会判断Type是否以"Cache.memcached"开头，如果是，则认为这是一个缓存操作，(这里代码我认为有些稍稍不合理，如果我用的是Redis缓存，我希望上报的Type="Cache.Redis"，所以我这里讲源码稍稍做了修改，判断Type如果以"Cache."开头，就认为是缓存)。

@RunWith(JUnit4.class)
public class AppSimulator extends CatTestCase {
    @Test
    public void simulateHierarchyTransaction() throws Exception {
        ...
        Transaction cacheT = cat.newTransaction("Cache.memcached", "get:uid_1234567");

        //do your cache operation

        cat.logEvent("Cache.memcached.server", "192.168.20.67:6379");
        cacheT.complete();
        ...
    }
}

　　

　　接下来我们看下Storage分析器的处理逻辑，processCacheTransaction负责分析消息， 事务类型"Cache.memcached"的“Cache.”后面部分将会被提取作为缓存类型，分析器会为每个类型的缓存都创建一个报表，事务名称":"前面部分会被提取作为操作名称，一般缓存有 add,get,hGet,mGet,remove等操作，缓存地址将由type="Cache.memcached.server"的Event子消息提供，最后我们还是将domain、ip、method、事务、阈值等消息放入StorageUpdateParam交由StorageReportUpdater来更新报表，更新逻辑与数据库一致。

public class StorageAnalyzer extends AbstractMessageAnalyzer<StorageReport> implements LogEnabled {
    @Inject
    private StorageReportUpdater m_updater;

    private void processCacheTransaction(MessageTree tree, Transaction t) {
        String cachePrefix = "Cache.";
        String ip = "Default";
        String domain = tree.getDomain();
        String cacheType = t.getType().substring(cachePrefix.length());
        String name = t.getName();
        String method = name.substring(name.lastIndexOf(":") + 1);
        List<Message> messages = t.getChildren();

        for (Message message : messages) {
            if (message instanceof Event) {
                String type = message.getType();

                if (type.equals("Cache.memcached.server")) {
                    ip = message.getName();
                    int index = ip.indexOf(":");

                    if (index > -1) {
                        ip = ip.substring(0, index);
                    }
                }
            }
        }
        String id = queryCacheId(cacheType);
        StorageReport report = m_reportManager.getHourlyReport(getStartTime(), id, true);
        StorageUpdateParam param = new StorageUpdateParam();

        param.setId(id).setDomain(domain).setIp(ip).setMethod(method).setTransaction(t)
              .setThreshold(LONG_CACHE_THRESHOLD);
        m_updater.updateStorageReport(report, param);
    }
}

StateAnalyzer

主要是分析CAT服务器自身的异常，他在周期任务运行中，不搜集任何数据，而是在周期结束后，对CAT的整体状况做一个汇总后生成报表，他的报表结构如下。

HeartbeatAnalyzer

分析器HeartbeatAnalyzer用于上报的心跳数据的分析。我们先看看客户端的收集逻辑，CAT客户端在初始化CatClientModule的时候，会开启一个StatusUpdateTask的线程任务，每隔一分钟去收集客户端的心跳状态，通过 Heartbeat 消息上报到客户端，心跳数据以xml格式存在于Heartbeat消息中。

public class CatClientModule extends AbstractModule {
    @Override
    protected void execute(final ModuleContext ctx) throws Exception {
        ...
        if (clientConfigManager.isCatEnabled()) {
            // start status update task
            StatusUpdateTask statusUpdateTask = ctx.lookup(StatusUpdateTask.class);

            Threads.forGroup("cat").start(statusUpdateTask);
            ...
        }
    }
}

Cat客户端会为Heartbeat消息创建一个System类型事务消息，然后将 Heartbeat 消息放入该事务，信息的收集靠StatusInfoCollector来完成，StatusInfoCollector将收集的数据写入StatusInfo对象，然后StatusUpdateTask将StatusInfo转化成xml之后放到Heartbeat消息数据段上报。

public class StatusUpdateTask implements Task, Initializable {
    @Override
    public void run() {
        //创建类目录, 上报CAT客户端启动信息
        ...

        while (m_active) {
            long start = MilliSecondTimer.currentTimeMillis();

            if (m_manager.isCatEnabled()) {
                Transaction t = cat.newTransaction("System", "Status");
                Heartbeat h = cat.newHeartbeat("Heartbeat", m_ipAddress);
                StatusInfo status = new StatusInfo();

                t.addData("dumpLocked", m_manager.isDumpLocked());
                try {
                    StatusInfoCollector statusInfoCollector = new StatusInfoCollector(m_statistics, m_jars);

                    status.accept(statusInfoCollector.setDumpLocked(m_manager.isDumpLocked()));

                    buildExtensionData(status);
                    h.addData(status.toString());
                    h.setStatus(Message.SUCCESS);
                } catch (Throwable e) {
                    h.setStatus(e);
                    cat.logError(e);
                } finally {
                    h.complete();
                }
                t.setStatus(Message.SUCCESS);
                t.complete();
            }

            //sleep 等待下一次心跳上报
            ...
        }
    }
}

我们上报的XML到底包含哪些数据，我们看下StatusInfo的结构，StatusInfo除了包含上报时间戳之外，还有哪些系统状态信息、附加扩展信息(Extension)会被StatusInfoCollector收集？

1、运行时数据RuntimeInfo：JAVA版本 java.version、用户名user.name、项目目录user.dir、java类路径等等。

2、操作系统信息 OsInfo，同时创建System附加扩展信息。

3、磁盘信息DiskInfo，磁盘的总量、空闲与使用情况，同时创建Disk附加扩展信息

4、内存使用情况MemoryInfo，同时创建垃圾回收扩展信息、JAVA虚拟机堆附加扩展信息

5、线程信息，以及 FrameworkThread 附加扩展信息。

6、Cat使用状态信息

下面我列一个上报的XML数据案例：

<?xml version="1.0" encoding="utf-8"?>
<status timestamp="2018-05-28 16:23:08.625">
   <runtime start-time="1527495705011" up-time="114212" java-version="1.8.0_40" user-name="CAOHAO1">
      <user-dir>E:\cat\cat-client</user-dir>
      <java-classpath>idea_rt.jar,junit-rt.jar,charsets.jar ... </java-classpath>
   </runtime>
   <os name="Windows 7" arch="amd64" version="6.1" available-processors="4" system-load-average="-1.0" process-time="5881237700" total-physical-memory="8538804224" free-physical-memory="1369870336" committed-virtual-memory="277372928" total-swap-space="17075703808" free-swap-space="4815220736"/>

   <disk>
      <disk-volume id="C:\" total="77218181120" free="23651565568" usable="23651565568"/>
      <disk-volume id="D:\" total="461373435904" free="249596563456" usable="249596563456"/>
      ...
   </disk>

   <memory max="1897922560" total="128974848" free="118387592" heap-usage="10961920" non-heap-usage="17081736">
      <gc name="PS Scavenge" count="1" time="167"/>
      <gc name="PS MarkSweep" count="0" time="0"/>
   </memory>
   <thread count="11" daemon-count="10" peek-count="11" total-started-count="12" cat-thread-count="0" pigeon-thread-count="0" http-thread-count="0">
      <dump>1: "Attach Listener" Id=5 RUNNABLE ... </dump>
   </thread>
   <message produced="0" overflowed="0" bytes="0"/>
   <extension id="System">
      <extensionDetail id="LoadAverage" value="-1.0"/>
      <extensionDetail id="FreePhysicalMemory" value="1.369247744E9"/>
      <extensionDetail id="FreeSwapSpaceSize" value="4.814426112E9"/>
   </extension>
   <extension id="Disk">
      <extensionDetail id="C:\ Free" value="2.3651565568E10"/>
      <extensionDetail id="D:\ Free" value="2.49596563456E11"/>
      ...
   </extension>
   <extension id="GC">
      <extensionDetail id="PS ScavengeCount" value="1.0"/>
      <extensionDetail id="PS ScavengeTime" value="167.0"/>
      <extensionDetail id="PS MarkSweepCount" value="0.0"/>
      <extensionDetail id="PS MarkSweepTime" value="0.0"/>
   </extension>
   <extension id="JVMHeap">
      <extensionDetail id="Code Cache" value="3707200.0"/>
      <extensionDetail id="Metaspace" value="1.2053E7"/>
      <extensionDetail id="Compressed Class Space" value="1412600.0"/>
      <extensionDetail id="PS Eden Space" value="4805792.0"/>
      <extensionDetail id="PS Survivor Space" value="5214992.0"/>
      <extensionDetail id="PS Old Gen" value="941136.0"/>
   </extension>
   <extension id="FrameworkThread">
      <extensionDetail id="HttpThread" value="0.0"/>
      <extensionDetail id="CatThread" value="0.0"/>
      <extensionDetail id="PigeonThread" value="0.0"/>
      <extensionDetail id="ActiveThread" value="11.0"/>
      <extensionDetail id="StartedThread" value="12.0"/>
   </extension>
   <extension id="CatUsage">
      <extensionDetail id="Produced" value="2.0"/>
      <extensionDetail id="Overflowed" value="0.0"/>
      <extensionDetail id="Bytes" value="1038.0"/>
   </extension>
</status>

我们再来看看服务端的分析逻辑，HeartbeatAnalyzer会为每个Domain创建一张心跳报表HeartbeatReport，不同IP的机器心跳数据存在于不同Machine对象里，每分钟的心跳数据都由一个Period对象存储；

HeartbeatAnalyzer首先将XML还原为StatusInfo，然后会用StatusInfo的RuntimeInfo、OsInfo、DiskInfo、MemoryInfo、ThreadInfo、MessageInfo的信息以及Extensions的动态属性m_dynamicAttributes去更新Period的m_extensions。

DumpAnalyzer -- 原始消息LogView存储

DumpAnalyzer 与其它分析器有点不同，它不是为了报表而设计，而是用于原始消息LogView的存储，与报表统计不一样，他的数据量非常大，几年前美团点评每天处理消息就达到1000亿左右，大小大约100TB，单物理机高峰期每秒要处理100MB左右的流量，因为数据量比较大所以存储整体要求就是批量压缩以及随机读，采用队列化、异步化、线程池等技术来保证并发。

当有客户端消息过来，DumpAnalyzer会调用本地消息处理器管理类(LocalMessageBucketManager) 的 storeMessage 方法存储消息，LocalMessageBucketManager是LogView管理的核心类，我们先看一看 LocalMessageBucketManager 对象的初始化函数 initialize() 的处理逻辑：

1、首先获取消息存储的基础路径(m_baseDir)，默认是 /data/appdatas/cat/bucket/dump， 在 server.xml 中可以配置，消息在基础路径之内，会根据domain、机器、时间等元素来分门别类的存储。

2、开启 BlockDumper 线程， 将本地消息处理器(LocalMessageBucket)、阻塞队列(BlockingQueue)以及统计信息的指针传入BlockDumper 对象，当内存消息块到达 64K 的时候， 该线程会异步将内存消息块写入数据文件和索引文件。

3、开启LogviewUploader线程，将自己的指针、本地消息处理器、HDFS上传对象(HdfsUploader)以及配置管理器的指针传入LogviewUploader对象，该用于异步将文件上传到 HDFS， 前提是配置了 hdfs 上传配置。

4、开启20个消息压缩线程(本地模式仅2个线程)，并为每个线程分配一个阻塞队列，当DumpAnalyzer接收到消息请求，会将消息写入该队列，MessageGzip会轮训从队列取消息处理，注意这里虽然有20个队列，然而正常我们只插入前19个队列，只有在前面入队失败了，消息将会被插入最后那个队列，可以认为最后那个队列是前面队列的一个备用队列。

public class LocalMessageBucketManager extends ContainerHolder implements MessageBucketManager, Initializable, LogEnabled {
    @Override
    public void initialize() throws InitializationException {
        m_baseDir = new File(m_configManager.getHdfsLocalBaseDir(ServerConfigManager.DUMP_DIR));

        Threads.forGroup("cat").start(new BlockDumper(m_buckets, m_messageBlocks, m_serverStateManager));
        Threads.forGroup("cat").start(new LogviewUploader(this, m_buckets, m_logviewUploader, m_configManager));

        if (m_configManager.isLocalMode()) {
            m_gzipThreads = 2;
        }

        for (int i = 0; i < m_gzipThreads; i++) {
            LinkedBlockingQueue<MessageItem> messageQueue = new LinkedBlockingQueue<MessageItem>(m_gzipMessageSize);

            m_messageQueues.put(i, messageQueue);
            Threads.forGroup("cat").start(new MessageGzip(messageQueue, i));
        }
        m_last = m_messageQueues.get(m_gzipThreads - 1);
    }

    @Override
    public void storeMessage(final MessageTree tree, final MessageId id) {
        boolean errorFlag = true;
        int hash = Math.abs((id.getDomain() + '-' + id.getIpAddress()).hashCode());
        int index = (int) (hash % m_gzipThreads);
        MessageItem item = new MessageItem(tree, id);
        LinkedBlockingQueue<MessageItem> queue = m_messageQueues.get(index % (m_gzipThreads - 1));
        boolean result = queue.offer(item);
        ...
    }
}

当DumpAnalyzer接收到消息请求，会调用storeMessage(...) 函数处理消息，如上源码，函数会根据domain和客户端ip将消息均匀分配到那19个阻塞队列(LinkedBlockingQueue)中，然后MessageGzip会轮询从队列获取消息数据，调用gzipMessage(item)函数处理，每处理 10000 条消息，MessageGzip会上报一条Gzip压缩线程监控记录。

我们再看看最核心的 gzipMessage(MessageItem item) 函数的处理逻辑，CAT根据日期，周期小时，domain，客户端地址，服务端地址创建存储路径和文件，包含数据文件和索引文件， 例如 20180611/15/Cat-127.0.01-127.0.01、20180611/15/Cat-127.0.01-127.0.01.idx ，从前面可以看出Message-ID的前3段可以确定唯一的索引文件，每条消息的存储由本地消息处理器(LocalMessageBucket)控制，LocalMessageBucket 的 storeMessage(...)方法会将消息信息写入消息块(MessageBlock)对象存放在内存中，当MessageBlock数据块大小达到 64K 时，将内存数据(MessageBlock) 放入阻塞队列 (m_messageBlocks)，异步写入文件，并清空内存MessageBlock。LocalMessageBucket 有个字段 m_blockSize 用于记录消息块总大小，注意这里的 64K 是压缩前的消息块总大小。

public class MessageGzip implements Task {
    private void gzipMessage(MessageItem item) {
        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);
            }
        }
    }
}

从上代码可以看出，当 storeMessage(...) 返回不为空的消息块(MessageBlock)时，则认为内存数据已经达到64K，需要写入文件，MessageGzip将消息块推入阻塞队列m_messageBlocks， BlockDumper线程会对队列进行消费， 它在实例化的时候会创建一个执行线程池 m_executors，然后 BlockDumper 线程轮询从阻塞队列取消息块(MessageBlock)，为每个消息块创建一个块写入任务(FlushBlockTask)，并将任务提交给执行线程池执行。FlushBlockTask实际会调用BlockDumper的flushBlock(block)函数将MessageBlock写入文件。

最终写入操作，还是得由LocalMessageBucket的MessageBlockWriter来完成，接下来我们介绍下本地消息处理器(LocalMessageBucket)，它是一个控制消息数据读写的对象，数据在内存中的载体是消息块(MessageBlock)，LocalMessageBucket 在gzipMessage(...)被首次实例化、初始化，初始化过程中会创建一个消息块(MessageBlock)、消息块读处理对象(MessageBlockReader)、消息块写处理对象(MessageBlockWriter)、、缓冲区以及缓冲区压缩流。

MessageBlock 包含4个信息：文件路径、数据缓冲区、每条ID的序列号、每条消息数据的大小(压缩前)。

消息块读处理对象负责消息的读取操作。

消息块写处理对象则负责数据文件、索引文件的写入操作，他会维护一个文件游标偏移量，记录压缩消息块(MessageBlock)在数据文件中的起始位置，即图2中块地址，下面是具体的写逻辑，先写索引文件，CAT先获取消息块中消息总条数，为每个Message-ID都写一个索引记录，每条消息的索引记录长度都是48bits，索引根据Message-ID的第四段（序列号）来确定索引的位置，比如消息Message-ID为ShopWeb-0a010680-375030-2，这条消息ID对应的索引位置为2*48bits的位置，48bits索引包含32bits的块地址 和 16bits 的块内偏移地址，前者记录压缩消息块(MessageBlock)在数据文件中的偏移位置，由于消息块包含多条消息，我们需要16bits来记录消息在消息块中的位置，注意这里指解压后的消息块。写完索引文件再写入数据文件，每一段压缩数据，前4位都是压缩块的大小，后面才是消息块的实际数据。

public class MessageBlockWriter {
    private RandomAccessFile m_indexFile;
    private RandomAccessFile m_dataFile;
    private int m_blockAddress;

    public synchronized void writeBlock(MessageBlock block) throws IOException {
        int len = block.getBlockSize();
        byte[] data = block.getData();
        int blockSize = 0;

        for (int i = 0; i < len; i++) {
            int seq = block.getIndex(i);
            int size = block.getSize(i);

            m_indexFile.seek(seq * 6L);
            m_indexFile.writeInt(m_blockAddress);
            m_indexFile.writeShort(blockSize);
            blockSize += size;
        }

        m_dataFile.writeInt(data.length);
        m_dataFile.write(data);
        m_blockAddress += data.length + 4;
    }
}

CAT读取消息的时候，首先根据Message-ID的前面三段确定唯一的索引文件，在根据Message-ID第四段确定此Message-ID索引位置，根据索引文件的48bits读取数据文件的内容，然后将数据文件进行GZIP解压，在根据块内偏移地址读取出真正的消息内容。

一定得注意的是，同一台客户端机器产生的Message-ID的第四段，即当前小时的顺序递增号，在当前小时内一定不能重复，因为在服务端，CAT会为每个客户端IP、每个小时的原始消息存储都创建一个索引文件，每条消息的索引记录在索引文件内的偏移位置是由顺序递增号决定的，一旦顺序号重复生成，那么该小时的重复索引数据将会被覆盖，导致我们无法通过索引找到原始消息数据。

上传HDFS

自定义分析器与报表