深刻详解美团点评CAT跨语言服务监控（四）服务端消息分发

时间  2019-11-12
标签 深刻 详解 点评 cat 跨语言 服务 监控 消息 分发 繁體版
原文   原文链接

这边首先介绍下大众点评CAT消息分发大概的架构以下：

 

图4 消息分发架构图算法

 

分析管理器的初始化服务器

 

    咱们在第一章讲到服务器将接收到的消息交给解码器(MessageDecoder)去作解码最后交给具体的消费者(RealtimeConsumer)去消费消息。架构

    RealtimeConsumer 是在何时被建立初始化？ 在第一章咱们讲到，CatHomeModule经过调用setup安装完成以后，会调用 execute 进行初始化的工做， 在execute方法中调用ctx.lookup(MessageConsumer.class) 方法来经过容器实例化RealtimeConsumer。dom

    在消费者中，最重要的一个概念就是消息分析器(MessageAnalyzer)，全部的消息分析统计，报表建立都是由消息分析器来完成，全部的分析器(MessageAnalyzer)都由消息分析器管理对象(MessageAnalyzerManager)管理，RealtimeConsumer就拥有消息分析器管理对象的指针，在消费者初始化以前，咱们会先实例化 MessageAnalyzerManager，而后调用initialize() 方法初始化分析管理器。异步

public class DefaultMessageAnalyzerManager extends ContainerHolder implements MessageAnalyzerManager, Initializable, LogEnabled {
    private List<String> m_analyzerNames;
    private Map<Long, Map<String, List<MessageAnalyzer>>> m_analyzers = new HashMap<Long, Map<String, List<MessageAnalyzer>>>();
    
    @Override
    public void initialize() throws InitializationException {
        Map<String, MessageAnalyzer> map = lookupMap(MessageAnalyzer.class);
        for (MessageAnalyzer analyzer : map.values()) {
            analyzer.destroy();
        }
 
        m_analyzerNames = new ArrayList<String>(map.keySet());
        ...
    }
}

 

    initialize() 方法经过IOC容器的lookupMap方法，找到全部的消息分析器。一共12个，以下图，而后取出分析器的名字，放到m_analyzerNames 列表里，能够认为每一个名字对应一种分析器，不一样的分析器都将从不一样角度去分析、统计上报的消息，汇总以后生成不一样的报表，咱们若是有本身的扩展需求,须要对消息作其它处理，也能够添加本身的分析器，只须要符合CAT准则便可。
async

 

消费者与周期管理器的初始化ide

消息分析器管理对象初始化以后，RealtimeConsumer 会执行 initialize() 来实现自身的初始化，函数

public class RealtimeConsumer extends ContainerHolder implements MessageConsumer, Initializable, LogEnabled {
    @Inject
    private MessageAnalyzerManager m_analyzerManager;
    
    private PeriodManager m_periodManager;
 
    @Override
    public void initialize() throws InitializationException {
        m_periodManager = new PeriodManager(HOUR, m_analyzerManager, m_serverStateManager, m_logger);
        m_periodManager.init();
 
        Threads.forGroup("cat").start(m_periodManager);
    }
}

    RealtimeConsumer的初始化很简单，仅包含3行代码，它的任务就是实例化并初始化周期管理器(PeriodManager)，并将分析器管理对象(MessageAnalyzerManager)的指针传给它，PeriodManager保留分析管理器指针仅仅用于在启动一个Period的时候，将MessageAnalyzerManager的指针传递给Period。spa

    PeriodManager的构造函数中，最核心的工做就是建立一个周期策略对象(PeriodStrategy)，每一个周期的开始/结束会参考PeriodStrategy的计算结果，变量duration是每一个周期的长度，默认是1个小时，并且周期时间是整点时段，例如：1:00-2:00, 2:00-3:00，周期时间是报表的最小统计单元，即分析器产生的每一个报表对象，都是当前周期时间内的统计信息。线程

    接下来RealtimeConsumer将会调用 m_periodManager.init() 启动第一个周期，仍是上面代码，咱们会计算当前时间所处的周期的开始时间，是当前时间的整点时间，好比如今是 13:50， 那么startTime=13:00，而后entTime=startTime + duration 算得结束时间为 14:00， 而后根据起始结束时间来建立 Period 对象，传入分析器的指针。并将周期对象加入到m_periods列表交给PeriodManager管理。最后调用period.start 启动第一个周期。

 

public class PeriodManager implements Task {
    private PeriodStrategy m_strategy;
    
    private List<Period> m_periods = new ArrayList<Period>();
    
    public PeriodManager(long duration, MessageAnalyzerManager analyzerManager,
          ServerStatisticManager serverStateManager, Logger logger) {
        m_strategy = new PeriodStrategy(duration, EXTRATIME, EXTRATIME);
        m_active = true;
        m_analyzerManager = analyzerManager;
        m_serverStateManager = serverStateManager;
        m_logger = logger;
    }
    
    public void init() {
        long startTime = m_strategy.next(System.currentTimeMillis());
 
        startPeriod(startTime);
    }
    
    private void startPeriod(long startTime) {
        long endTime = startTime + m_strategy.getDuration();
        Period period = new Period(startTime, endTime, m_analyzerManager, m_serverStateManager, m_logger);
 
        m_periods.add(period);
        period.start();
    }
}

 

 

    咱们再回到ReatimeConsumer的initialize()初始化方法，第三行代码，Threads.forGroup("cat").start(m_periodManager) 将开启一个周期管理线程，线程执行代码以下run()函数，每隔1秒钟会计算是否须要开启一个新的周期，value>0就开启新的周期， value=0啥也不干，value<0的异步开启一个新线程结束上一个周期。结束线程调用PeriodManager的endPeriod(long startTime)方法完成周期的清理工做，而后将period从m_periods列表移除出去。

public class PeriodManager implements Task {
    private List<Period> m_periods = new ArrayList<Period>();
 
    @Override
    public void run() {
        while (m_active) {
            try {
                long now = System.currentTimeMillis();
                long value = m_strategy.next(now);
 
                if (value > 0) {
                    startPeriod(value);
                } else if (value < 0) {
                    // last period is over,make it asynchronous
                    Threads.forGroup("cat").start(new EndTaskThread(-value));
                }
            } catch (Throwable e) {
                Cat.logError(e);
            }
            
            Thread.sleep(1000L);
        }
    }
    
    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;
            }
        }
    }
}

 

 

什么是周期？

    好了，咱们在上两节讲了分析器的初始化，周期管理器的初始化，那么，什么是周期？为何会有周期？他是如何工做的？

    能够认为周期Period就是一个消息分发的控制器，至关于MVC的Controller，受PeriodManager的管理，全部客户端过来的消息，都会根据消息时间戳从PeriodManager中找到消息所属的周期对象(Period)，由该周期对象来派发消息给每一个注册的分析器(MessageAnalyzer)来对消息作具体的处理。

    然而Period并非直接对接分析器(MessageAnalyzer)， 而是经过PeriodTask来与MessageAnalyzer交互，Period类有个成员变量m_tasks， 类型为Map<String, List<PeriodTask>>， Map的key是String类型，表示分析器的名字，好比top、cross、transaction、event等等，咱们一共有12种类别的分析器，不过实际处理过程当中，CAT做者移除了他认为比较鸡肋的Matrix、Dependency两个分析器，只剩下10个分析器了，如图10。

m_analyzerNames.remove("matrix");
m_analyzerNames.remove("dependency");

图10：参与任务处理的分析器名称

    Map的value为List<PeriodTask> 是一个周期任务的列表， 也就是说，每一种类别的分析器，都会有至少一个MessageAnalyzer的实例，每一个MessageAnalyzer都由一个对应的PeriodTask来分配任务，MessageAnalyzer与PeriodTask是1对1的关系，每种类别分析器具体有多少个实例由 getAnalyzerCount() 函数决定，默认是 1 个， 可是有些分析任务很是耗时，须要多个线程来处理，保证处理效率，好比 TransactionAnalyzer就是2个。

public class TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionReport> implements LogEnabled {
    @Override
    public int getAnalyzerCount() {
        return 2;
    }
}

 

    消息分发的时候，每一笔消息默认都会发送到全部种类分析器处理，可是同一种类别的分析器下若是有多个MessageAnalyzer实例，采用domain hash 选出其中一个实例安排处理消息，分发算法参考下面源码：

public class Period {
    private Map<String, List<PeriodTask>> m_tasks;
    
    public void distribute(MessageTree tree) {
        ...
        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;
            }
            PeriodTask task = tasks.get(index);
            boolean enqueue = task.enqueue(tree);
            ...
        }
        ...
    }
}

 

 

周期任务-任务队列

    上一节咱们讲到与MessageAnalyzer交互是由PeriodTask来完成的，那么周期任务PeriodTask在哪里被建立？他怎么与分析器进行交互， 在Period实例化的同时，PeriodTask就被建立了，咱们看看Period类的构造函数：

public class Period {
    private Map<String, List<PeriodTask>> m_tasks;
    
    public Period(long startTime, long endTime, MessageAnalyzerManager analyzerManager,
          ServerStatisticManager serverStateManager, Logger logger) {
          
        ...
        
        List<String> names = m_analyzerManager.getAnalyzerNames();
 
        m_tasks = new HashMap<String, List<PeriodTask>>();
        for (String name : names) {
            List<MessageAnalyzer> messageAnalyzers = m_analyzerManager.getAnalyzer(name, startTime);
 
            for (MessageAnalyzer analyzer : messageAnalyzers) {
                MessageQueue queue = new DefaultMessageQueue(QUEUE_SIZE);
                PeriodTask task = new PeriodTask(analyzer, queue, startTime);
                
                //加入 m_tasks
                ...
            }
        }
    }
}

 

 

    构造函数首先获取全部分析器名字，咱们说过每一个名字对应一种分析器，而后根据分析器名字和周期时间，获取当前周期、该种类分析器全部实例，以前说过，有些类别分析任务逻辑复杂，耗时长，会须要更多的分析线程处理，为每一个分析器都建立一个 PeriodTask，并为每个PeriodTask建立任务队列。客户端消息过来，会由Period分发给全部种类的PeriodTask，同一类分析器下有多个分析器(MessageAnalyzer)的时候，只有一个MessageAnalyzer会被分发，采用domain hash选出这个实例，在这里，分发实际上就是插入PeriodTask的任务队列。

    构造函数最后将建立PeriodTask加入m_tasks中。

    在Period被实例化以后， 周期管理器(PeriodManager)就调用 period.start() 开启一个周期了，start逻辑很简单， 就是启动period下全部周期任务(PeriodTask)线程。任务线程也很是简单，就是调用本身的分析器的分析函数analyze(m_queue)来处理消息。

public class PeriodTask implements Task, LogEnabled {
 
    private MessageAnalyzer m_analyzer;
 
    private MessageQueue m_queue;
    
    @Override
    public void run() {
        try {
            m_analyzer.analyze(m_queue);
        } catch (Exception e) {
            Cat.logError(e);
        }
    }
}

    接下来咱们看下分析函数作了什么，下面是源码，只展现了核心逻辑部分，分析程序轮训从PeriodTask传入的任务队列中取出消息，而后调用process处理，具体的处理逻辑就是由process完成的，process是一个抽象函数，具体实现由各类类分析器子类来实现，咱们将在下一章分别讲解。

    固然这里的前提是分析器处在激活状态，而且本周期未结束，结束的定义是当前时间比周期时间+延迟结束时间更晚，延迟结束时间会在后面周期策略章节详细讲解，一旦周期结束，分析器将会把剩余的消息分析完而后关闭。

public abstract class AbstractMessageAnalyzer<R> extends ContainerHolder implements MessageAnalyzer {
    protected abstract void process(MessageTree tree);
 
    @Override
    public void analyze(MessageQueue queue) {
        while (!isTimeout() && isActive()) {
            MessageTree tree = queue.poll();
 
            if (tree != null) {
                ...
                process(tree);
                ...
            }
        }
        ...
    }
    
    protected boolean isTimeout() {
        long currentTime = System.currentTimeMillis();
        long endTime = m_startTime + m_duration + m_extraTime;
 
        return currentTime > endTime;
    }
}

 

 

消息分发

 

消息从客户端发上来，是如何到达PeriodTask的，又是如何分配分析器的？

客户端消息发送到服务端，通过解码以后，就调用 MessageConsumer的 consume() 函数对消息进行消费。源码以下：

public class RealtimeConsumer extends ContainerHolder implements MessageConsumer, Initializable, LogEnabled {
    @Override
    public void consume(MessageTree tree) {
        String domain = tree.getDomain();
        String ip = tree.getIpAddress();
 
        if (!m_blackListManager.isBlack(domain, ip)) {
            long timestamp = tree.getMessage().getTimestamp();
            Period period = m_periodManager.findPeriod(timestamp);
 
            if (period != null) {
                period.distribute(tree);
            } else {
                m_serverStateManager.addNetworkTimeError(1);
            }
        } else {
            m_black++;
 
            if (m_black % CatConstants.SUCCESS_COUNT == 0) {
                Cat.logEvent("Discard", domain);
            }
        }
    }
}

 

 

    consume函数会首先判断domain和ip是否黑名单，若是是黑名单，丢弃消息，不然，根据消息时间戳，找到对应的周期(Period)，交给Period对消息进行分发，分发逻辑前面讲过，Period将消息插入PeriodTask队列，由分析器(MessageAnalyzer)轮训从队列里面取消息进行具体处理，每笔消息默认会被全部类别分析器处理，当同一类别分析器有多个MessageAnalyzer实例的时候，选择其中一个处理，选择算法：

Math.abs(domain.hashCode()) % length

 

 

详细的源码可参考章节什么是周期？

 

周期策略

    在建立周期策略对象的时候，会传入3个参数，一个是duration，也就是每一个周期的时间长度，默认为1个小时，另外两个extraTime和aheadTime分别表示我提早启动一个周期的时间和延迟结束一个周期的时间，默认都是3分钟，咱们并不会卡在整点时间，例如10:00去开启或结束一个周期，由于周期建立是须要消耗必定时间，这样能够避免消息过来周期对象还未建立好，或者消息尚未处理完，就要去结束周期。

    固然，即便提早建立了周期对象(Period)，并不意味着就会当即被分发消息，只有到了该周期时间才会被分发消息。

    下面看看具体的策略方法，咱们首先计算当前时间的周期启动时间(startTime)，是当前时间的整点时间，好比当前时间是 22:47.123，那么startTime就是 22:00.000，注意这里的时间都是时间戳，单位为毫秒。

    接下来判断是否开启当前周期，若是startTime大于上次周期启动时间(m_lastStartTime)，说明应该开启新的周期，因为m_lastStartTime初始化为 -1， 因此CAT服务端初始化以后第一个周期会执行到这里，并记录m_lastStartTime。

    上面if若是未执行，咱们会判断当前时间比起上次周期启动时间是否是已通过了 57 分钟(duration - aheadTime )，即提早3分钟启动下一个周期。

    若是上面if还未执行，咱们则认为当前周期已经被启动，那么会判断是否须要结束当前周期，即当前时间比起上次周期启动时间是否是已通过了 63 分钟(duration + extraTime)，即延迟3分钟关闭上一个周期。

public class PeriodStrategy {
    public long next(long now) {
        long startTime = now - now % m_duration;
 
        // for current period
        if (startTime > m_lastStartTime) {
            m_lastStartTime = startTime;
            return startTime;
        }
 
        // prepare next period ahead
        if (now - m_lastStartTime >= m_duration - m_aheadTime) {
            m_lastStartTime = startTime + m_duration;
            return startTime + m_duration;
        }
 
        // last period is over
        if (now - m_lastEndTime >= m_duration + m_extraTime) {
            long lastEndTime = m_lastEndTime;
            m_lastEndTime = startTime;
            return -lastEndTime;
        }
 
        return 0;
    }
}
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程