深度解析Cat源码系列专栏点击访问

CAT源码分析5 - 服务端消息接收

1. TcpSocketReceiver初始化

在CAT服务端启动初始化过程中，CatHomeModule在初始化时，会通过setUp()和execute()两个方法实现对消息接收器的初始化。

• 通过setUp方法完成对Netty客户端网络通信的初始化
• execute方法中有其他功能的初始化逻辑，包括对消息消费者的初始化

整体的初始化逻辑如下图

com.dianping.cat.CatHomeModule#setup 设置方法如下

• 通过Peluxs容器实例化TcpSocketReceiver，并执行init初始化
• 对进程设置销毁监听

@Override
protected void setup(ModuleContext ctx) throws Exception {final TcpSocketReceiver messageReceiver = ctx.lookup(TcpSocketReceiver.class);messageReceiver.init();// 销毁监听Runtime.getRuntime().addShutdownHook(new Thread() {@Overridepublic void run() {messageReceiver.destory();}});
}

com.dianping.cat.analysis.TcpSocketReceiver#init的初始化逻辑如下：

• 主要负责实例化Netty客户端，开启对2280端口通信的监听
• 选择网络通信方式时，根据操作系统，linux下使用epoll更加高效
• 绑定消息编解码器、设置网络配置参数
• m_logger是通过Plexus容器的LoggerManager自动初始化注入的，LoggerManager默认设置为Log4j。该logger的打印是落磁盘的

public void init() {try {// 默认2280端口启动Netty客户端startServer(m_port);} catch (Exception e) {m_logger.error(e.getMessage(), e);}
}public synchronized void startServer(int port) throws InterruptedException {// 根据操作系统，设置netty的相关工作线程数量和channel通信方式boolean linux = getOSMatches("Linux") || getOSMatches("LINUX");int threads = 24;ServerBootstrap bootstrap = new ServerBootstrap();m_bossGroup = linux ? new EpollEventLoopGroup(threads) : new NioEventLoopGroup(threads);m_workerGroup = linux ? new EpollEventLoopGroup(threads) : new NioEventLoopGroup(threads);bootstrap.group(m_bossGroup, m_workerGroup);// linux使用epollbootstrap.channel(linux ? EpollServerSocketChannel.class : NioServerSocketChannel.class);bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();// pipeline添加消息编解码器pipeline.addLast("decode", new MessageDecoder());pipeline.addLast("encode", new ClientMessageEncoder());}});// 对bootstrap进行网络通信配置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);}
}

2. MessageDecoder消息解码器

MessageDecoder继承了ByteToMessageDecoder类，在netty通道上有消息进入时会被调用。对Byte字节码进行解析。且，MessageDecoder是TcpSocketReceiver的一个内部类，直接引用了后者注入的MessageHandler和ServerStatisticManager，是服务端消息处理的入口。

在其代码中，首先对消息定长读取，获取消息的长度，实现对半包/粘包问题的处理。之后，读到消息体，通过NativeMessageCodec或PlainTextMessageCodec消息解码实现类，将byte转换为MessageTree对象。再使用MessageHandler实现对消息的处理消费。最后，通过ServerStatisticManager对处理消息情况进行统计。

public class MessageDecoder extends ByteToMessageDecoder {// 已经处理的消息数量private long m_processCount;@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {// 定长解码，头4字节是消息体长度if (buffer.readableBytes() < 4) {// 不足4字节，半包return;}buffer.markReaderIndex();int length = buffer.readInt();buffer.resetReaderIndex();if (buffer.readableBytes() < length + 4) {// 小于length半包return;}try {if (length > 0) {// 定长读取，避免粘包ByteBuf readBytes = buffer.readBytes(length + 4);readBytes.markReaderIndex();// 读取数据进行解码，默认是NativeMessageCodec，实现逻辑就是各种长度协议，不详述了DefaultMessageTree tree = (DefaultMessageTree) CodecHandler.decode(readBytes);readBytes.resetReaderIndex();tree.setBuffer(readBytes);// 消息处理器，准备消费消息m_handler.handle(tree);// 统计数量，每1000条执行一次统计m_processCount++;long flag = m_processCount % CatConstants.SUCCESS_COUNT;if (flag == 0) {m_serverStateManager.addMessageTotal(CatConstants.SUCCESS_COUNT);}} else {// client message is errorbuffer.readBytes(length);BufReleaseHelper.release(buffer);}} catch (Exception e) {m_serverStateManager.addMessageTotalLoss(1);m_logger.error(e.getMessage(), e);}}
}

3. MessageConsumer消费

3.1 消费机制

CAT服务端的消息消费者并没有直接对所有收到的消息进行计算等逻辑，而是使用“生产消费者”模式，异步的处理消息，以此避免阻塞提高消息的处理效率。接收消息作为生产，以小时为单位划分Period进行消费管理。

为什么要按小时生成Period来充当消费者？

1. 由于CAT是内存实时计算，按小时划分可以有效节省内存，某小时内消息异常，不会影响下一小时的计算，提高可靠性
2. 按小时处理消息，只处理当前一小时的数据，避免堆积造成内存泄漏

3.2 Period任务管理

MessageConsumer通过PeriodManager对周期进行管理，通过周期策略PeriodStrategy类创建/删除Period。Period管理所有的PeriodTask和消息分析器Analyser，实现对消息的处理。他们的关系如下

消息分析器MessageAnalyzer具有多种，如上图所示，通过plexus容器注入，后续介绍。

3.2.1 RealtimeConsumer

RealtimeConsumer注入了PeriodManger，前者的初始化方法，完成了对PeriodManger的初始化，并启动线程来循环执行其管理任务。com.dianping.cat.analysis.RealtimeConsumer#initialize()是逻辑入口，代码如下：

@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);//初始化过滤配置/filter.properties,排除不需要的domain对应的消息ExcludeConfigInitializer.initialize();
}

3.2.2 PeriodManager

PeriodManager继承自Task，不断轮训是否需要生成新的，或删除过期的Period，并开启Period。PeriodStrategy策略用来判断创建/删除。
PeriodManager在实例化时，会创建PeriodStrategy设置默认的周期1Hour，提前、超时时间都是3min。并在init方法初始化时，开启第一个period周期。
PeriodStrategy用于计算下一个周期的开始时间，当返回值大于0时，则PeriodManager创建新的Period；小于0时删除过期的Period；等于0时，还在当前周期，不需要创建和删除。

public PeriodManager(long duration, MessageAnalyzerManager analyzerManager,	ServerStatisticManager serverStateManager,Logger logger) {// 实例化周期创建策略 1小时为周期，ahead和expire都是3minm_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);
}@Override
public void run() {while (m_active) {try {long now = System.currentTimeMillis();// 计算下次开启Period的时间，大于0需要开启新的，小于0说明有过期需要删除long value = m_strategy.next(now);if (value > 0) {// 开启PeriodstartPeriod(value);} else if (value < 0) {// 上一个Period已经过期，异步卸载PeriodThreads.forGroup("cat").start(new EndTaskThread(-value));}} catch (Throwable e) {Cat.logError(e);}try {Thread.sleep(1000L);} catch (InterruptedException e) {break;}}
}

开启周期逻辑如下

private void startPeriod(long startTime) {long endTime = startTime + m_strategy.getDuration();Period period = new Period(startTime, endTime, m_analyzerManager, m_serverStateManager, m_logger);// 添加到集合中进行管理，通常就2个，当前小时Period + 不超过3min的LastPeriodm_periods.add(period);period.start();
}

删除逻辑是判断Period的开始时间是否相等，删除过期的Period

private class EndTaskThread implements Task {private long m_startTime;public EndTaskThread(long startTime) {m_startTime = startTime;}@Overridepublic String getName() {return "End-Consumer-Task";}@Overridepublic void run() {// 执行结束方法endPeriod(m_startTime);}
}
private void endPeriod(long startTime) {int len = m_periods.size();// 遍历peroid集合，结束删除相关周期for (int i = 0; i < len; i++) {Period period = m_periods.get(i);if (period.isIn(startTime)) {// 处理checkPoint，记录日志period.finish();// 移除周期m_periods.remove(i);break;}}
}

周期策略分析

PeriodStrategy的核心逻辑如下：

1. 通过当前时间毫秒数计now算出当前周期的开始时间startTime（小时整点）
2. startTime是否大于m_lastStartTime上一个周期的开始时间，初始化时，m_lastStartTime标记为0，故能创建。之后到57min之前，m_lastStartTime == 当前周期startTime，返回0，不创建新的Period
3. 当now超过57分（m_duration-m_aheadTime = 60 -3 = 57）后，标记m_lastStartTime为下一个小时的整点，并返回下一个小时的整点startTime + 1hour，提前3min创建下一个Period。
4. 之后，再次调用next时，now-m_lastEndTime由于m_lastEndTime初始化为0，差一定大于63。故，返回0，m_lastEndTime设置为当前周期的startTime
5. 之后的调用，3个判断都不满足，始终返回0，不创建新的period，也不删除period。直到now到达下一个小时，且超过3分钟时，返回上m_lastEndTime的负值，标记将对应Period删除

核心代码如下：

public long next(long now) {long startTime = now - now % m_duration;// m_lastStartTime是一个标记为，记录上一个周期的开始时间// 创建当前小时的periodif (startTime > m_lastStartTime) {m_lastStartTime = startTime;return startTime;}// 57分以后，计算出下一个小时的整点, 提前3min创建下个periodif (now - m_lastStartTime >= m_duration - m_aheadTime) {m_lastStartTime = startTime + m_duration;return startTime + m_duration;}// last period is over// 大于63, 上次结束时间更新为当前小时，返回负数，清理上一小时的periodif (now - m_lastEndTime >= m_duration + m_extraTime) {long lastEndTime = m_lastEndTime;m_lastEndTime = startTime;return -lastEndTime;}return 0;
}

3.3 Period&PeriodTask

Period通过PeriodManager创建和开启，实例时会创建所有的MessageAnalyzer的PeriodTask集合，通过Start方法开启后，遍历执行PeriodTask消费消息。PeriodTask种使用BlockingQueue来暂存Consumer生产的MessageTree消息，并从中读取进行消费。
Period的实例化逻辑：

• 通过AnalyzerManager获取配置的所有消息分析器名称，并创建对应的PeriodTask列表（分析器下次讲）
• 同一name的Analyzer实例默认会创建2个来进行消费，可配置（分析器下次讲）

public Period(long startTime, long endTime, MessageAnalyzerManager analyzerManager,ServerStatisticManager serverStateManager, Logger logger) {m_startTime = startTime;m_endTime = endTime;m_analyzerManager = analyzerManager;m_serverStateManager = serverStateManager;m_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) {// 通过name，分析器实例创建分析任务MessageQueue queue = new DefaultMessageQueue(QUEUE_SIZE);PeriodTask task = new PeriodTask(analyzer, queue, startTime);task.enableLogging(m_logger);List<PeriodTask> analyzerTasks = m_tasks.get(name);if (analyzerTasks == null) {analyzerTasks = new ArrayList<PeriodTask>();m_tasks.put(name, analyzerTasks);}analyzerTasks.add(task);}}
}

Period的Start方法，记录相关信息，通过线程开始遍历执行PeriodTask；distribute方法用于遍历所有管理的task，遍历并添加消息到队列中。

public void start() {// 记录信息SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");m_logger.info(String.format("Starting %s tasks in period [%s, %s]", m_tasks.size(),	df.format(new Date(m_startTime)),df.format(new Date(m_endTime - 1))));for (Entry<String, List<PeriodTask>> tasks : m_tasks.entrySet()) {List<PeriodTask> taskList = tasks.getValue();for (int i = 0; i < taskList.size(); i++) {PeriodTask task = taskList.get(i);task.setIndex(i);// 开启任务Threads.forGroup("Cat-RealtimeConsumer").start(task);}}
}// 上层生产消息是调用
public void distribute(MessageTree tree) {// 统计消息数量m_serverStateManager.addMessageTotal(tree.getDomain(), 1);boolean success = true;String domain = tree.getDomain();// 遍历所有的taskfor (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) {// 根据domain的hash值，选择一个task负载任务index = Math.abs(domain.hashCode()) % length;}PeriodTask task = tasks.get(index);// 添加到队列，失败时尝试向下一个task添加boolean enqueue = task.enqueue(tree);if (!enqueue) {if (manyTasks) {task = tasks.get((index + 1) % length);enqueue = task.enqueue(tree);if (!enqueue) {success = false;}} else {success = false;}}}
}

PeriodTask的核心方法如下，

• run方法调用分析器解析消息
• enqueue方法判断数据是否合格，并尝试添加到queue中

public boolean enqueue(MessageTree tree) {if (m_analyzer.isEligable(tree)) {// 尝试加入队列boolean result = m_queue.offer(tree);if (!result) { // 加入队列失败，1000条记录一次日志m_queueOverflow++;if (m_queueOverflow % (10 * CatConstants.ERROR_COUNT) == 0) {String date = new SimpleDateFormat("yyyy-MM-dd HH:mm").format(new Date(m_analyzer.getStartTime()));m_logger.warn(m_analyzer.getClass().getSimpleName() + " queue overflow number " + m_queueOverflow	+ " analyzer time:" + date);}}return result;} else {return true;}
}

3.4 RealtimeConsumer

RealtimeConsumer的consume方法在解码器中通过MessageHandler被调用。消费时，过滤特定的路径，并根据消息时间戳查找Period。找到Period后，执行distrube方法进行消息分发，即获取到所有的PeriodTask遍历执行enqueue，添加消息到队列中。
com.dianping.cat.analysis.RealtimeConsumer#consume

@Override
public void consume(MessageTree tree) {ExcludePathFilter pathFilter = ExcludeConfigInitializer.getPathFilter();String domain = tree.getDomain();// /filter.properties 过滤domainif (!pathFilter.filter(domain)) {long timestamp = tree.getMessage().getTimestamp();// 根据消息时间查找PeriodPeriod period = m_periodManager.findPeriod(timestamp);if (period != null) {// 分发消息period.distribute(tree);} else {m_serverStateManager.addNetworkTimeError(1);}} else {//记录错误 domain 数量Cat.logEvent("DroppedIllegalDomain",domain);}
}

整体的生产-消费流程如下图：