这边首先介绍下大众点评CAT消息分发大概的架构如下：
图4 消息分发架构图
分析管理器的初始化
我们在第一章讲到服务器将接收到的消息交给解码器(MessageDecoder)去做解码最后交给具体的消费者(RealtimeConsumer)去消费消息。
RealtimeConsumer 是在什么时候被创建初始化？ 在第一章我们讲到，CatHomeModule通过调用setup安装完成之后，会调用 execute 进行初始化的工作， 在execute方法中调用ctx.lookup(MessageConsumer.class) 方法来通过容器实例化RealtimeConsumer。
在消费者中，最重要的一个概念就是消息分析器(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准则即可。
消费者与周期管理器的初始化
消息分析器管理对象初始化之后，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。
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;
}
}