Cat模块
Cat-client : cat客户端，编译后生成 cat-client-2.0.0.jar ，用户可以通过它来向cat-home上报统一格式的日志信息，可以集成到 mybatis、spring、微服务 dubbo 的监控等等流行框架。
Cat-consumer： 用于实时分析从客户端提供的数据。在实际开发和部署中，Cat-consumer和Cat-home是部署在一个JVM内部，每个CAT服务端都可以作为consumer也可以作为home，这样既能减少整个层级结构，也可以增加系统稳定性。
Cat-core：Cat核心模块
Cat-hadoop ： 大数据统计依赖模块。
cat-home：大众点评CAT服务器端主程序，编译安装之后生成 cat-alpha-2.0.0.war 包部署于servlet容器中，我们用的是Tomcat，war包依赖cat-client.jar、cat-consumer.jar， cat-core.jar， cat-hadoop.jar 包，通过web.xml 配置，看到Cat会启动 cat-servlet 和 mvc-servlet , mvc-servlet 是一个类似 spring MVC 的框架，用于处理用户WEB管理平台请求。cat-servlet是CAT服务端监听入口，CAT会在这里开启监听端口，接收处理客户端的日志记录请求，本章主要介绍cat-servlet。
<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
version="2.5">
...
<servlet>
<servlet-name>cat-servlet</servlet-name>
<servlet-class>com.dianping.cat.servlet.CatServlet</servlet-class>
<load-on-startup>1</load-on-startup>
</servlet>
<servlet>
<servlet-name>mvc-servlet</servlet-name>
<servlet-class>org.unidal.web.MVC</servlet-class>
<init-param>
<param-name>cat-client-xml</param-name>
<param-value>client.xml</param-value>
</init-param>
<init-param>
<param-name>init-modules</param-name>
<param-value>false</param-value>
</init-param>
<load-on-startup>2</load-on-startup>
</servlet>
....
Cat-servlet初始化
图1 - 容器初始化类图
CatServlet 首先会调用父类 AbstractContainerServlet 的init方法做初始化工作， 可以认为这是CatServlet的入口，他主要做了3件事情，首先调用基类HttpServlet的init方法对Servlet进行初始化，然后初始化Plexus容器，最后调用子类initComponents初始化Module模块。
public abstract class AbstractContainerServlet extends HttpServlet {
public void init(ServletConfig config) throws ServletException {
super.init(config);
try {
if(this.m_container == null) {
this.m_container = ContainerLoader.getDefaultContainer();
}
this.m_logger = this.m_container.getLogger();
this.initComponents(config);
} catch (Exception var3) {
...
}
}
}
plexus - IOC容器
上面讲到init(...)方法在初始化完Servlet之后调用 ContainerLoader.getDefaultContainer() 初始化plexus容器。
注：这里可能大家不太了解plexus，它相当于Spring的IoC容器，但是它和Spring框架不同，它并不是一个完整的，拥有各种组件的大型框架，仅仅是一个纯粹的IoC容器，它的开发者与Maven的开发者是同一拨人，最初开发Maven的时候，Spring并不成熟，所以Maven的开发者决定使用自己维护的IoC容器Plexus，它与Spring在语法和描述方式稍有不同。在Plexus中，有ROLE的概念，相当于Spring中的一个Bean。支持组件生命周期管理。
非JAVA开发者不懂IOC容器？简单来说，IOC容器就相当于一个对象装载器，对象不是由程序员new创建，而是框架在初始化的时候从配置文件中读取需要实例化的类信息，将信息装入一个对象装载器，然后在需要的时候，从对象装载器中找是否存在该类的信息，存在则返回类的对象。
plexus容器是如何工作的呢？就上面的类图来说，
a. AbstractContainerServlet 通过容器工厂ContainerLoader 的 getDefaultContainer方法，该方法会创建 MyPlexusContainer 容器，MyPlexusContainer是接口 PlexusContainer 的实现，MyPlexusContainer在构造函数中会创建组件管理器(ComponentManager)，可以认为每个类都是容器中的一个组件，ComponentManager就是用来管理这些组件的，包括他的生命周期，组件在Plexus容器配置文件中配置。
b.组件管理器(ComponentManager)会创建组件模型管理器(ComponentModelManager)以及组件生命周期管理器(ComponentLifecycle)，ComponentModelManager用于存储Plexus容器配置文件中的所有component组件信息，它的loadComponentsFromClasspath()方法会扫描各个jar包中存在的plexus容器配置文件，如图2，将xml内容解析之后放入PlexusModel 列表中。
public class ComponentManager {
private Map<String, ComponentBox<?>> m_components = new HashMap();
private PlexusContainer m_container;
private ComponentLifecycle m_lifecycle;
private ComponentModelManager m_modelManager;
private LoggerManager m_loggerManager;
public ComponentManager(PlexusContainer container, InputStream in) throws Exception {
this.m_container = container;
this.m_modelManager = new ComponentModelManager();
this.m_lifecycle = new ComponentLifecycle(this);
if(in != null) {
this.m_modelManager.loadComponents(in);
}
this.m_modelManager.loadComponentsFromClasspath();
this.m_loggerManager = (LoggerManager)this.lookup(new ComponentKey(LoggerManager.class, (String)null));
this.register(new ComponentKey(PlexusContainer.class, (String)null), container);
this.register(new ComponentKey(Logger.class, (String)null), this.m_loggerManager.getLoggerForComponent(""));
}
}
我们也可以将我们自己写的类交给容器管理，只需要将类配置到容器配置文件中，例如：cat-consumer/src/main/resources/META-INF/plexus/components-cat-consumer.xml， 只要是存在于 META-INF/plexus/ 目录下，并且文件名以"components-" 开头的 ".xml" 文件，都会被 ComponentModelManager 认为是容器配置文件。
图2 - plexus IOC容器类配置文件
c.然后就可以通过lookup方法找到类，并在首次使用的时候实例化，并且xml配置中的该类依赖的其它类也会被一并实例化，另外如果类方法实现了 Initializable 接口，创建对象后会执行类的 initialize() 方法做一些初始化的工作。
if(component instanceof Initializable) {
try {
((Initializable)component).initialize();
} catch (Throwable var5) {
ComponentModel model = ctx.getComponentModel();
throw new ComponentLookupException("Error when initializing component!", model.getRole(), model.getHint(), var5);
}
}
模块的加载 - 模型模式
init(...)函数最后会调用CatServlet的initComponents()方法初始化Module模块。
图3 - 模块初始化类图
initComponents()方法首先创建一个模块上下文 DefaultModuleContext对象,该对象拥有plexus容器的指针，以及server.xml、client.xml配置文件信息 ，服务端配置server.xml中有消息存储路径、HDFS上传等一系列配置，由于cat-home默认是服务端也是客户端，也就是说cat-home自身也会被监控，所以我们在这里看到有client.xml配置，配置文件所在目录由环境变量CAT_HOME指定，如果未指定，默认是/data/appdatas/cat。
随后CatServlet创建一个模块初始器 DefaultModuleInitializer，并调用他的execute(ctx)方法创建并初始化模块。
注：DefaultModuleInitializer有一个模块管理器DefaultModelManager m_manager, 读者可能没有看见m_manager的创建过程，实际上，对象在components-foundation-service.xml配置文件中配置的，然后在plexus容器实例化类对象的过程中创建的，后面还有很多对象的属性也是通过plexus容器注入的。比如DefaultModuleManager的m_topLevelModules属性通过以下配置注入。
<component>
<role>org.unidal.initialization.ModuleManager</role>
<implementation>org.unidal.initialization.DefaultModuleManager</implementation>
<configuration>
<topLevelModules>cat-home</topLevelModules>
</configuration>
</component>
上面XML配置显示m_topLevelModules 指定为cat-home，这样DefaultModuleInitializer通过DefaultModelManager的getTopLevelModules()方法获取的就是CatHomeModule模块对象，可以认为cat-home是一个顶层模块，所有Module都包含getDependencies方法，该方法会找到当前模块所依赖的其他模块，并实例化模块，比如下面cat-home就依赖cat-consumer模块，
public class CatHomeModule extends AbstractModule {
@Override
public Module[] getDependencies(ModuleContext ctx) {
return ctx.getModules(CatConsumerModule.ID);
}
}
从cat-consumer的getDependencies看出他依赖cat-core模块，cat-core模块又依赖cat-client模块，这样子我们就从顶层模块引出了所有依赖的其它模块，在实例化模块的同时调用模块的setup方法安装模块。在所有模块安装完成之后，依次调用模块的execute方法完成初始化，但是初始化顺序则是按照安装顺序反着来的，cat-client -> cat-core -> cat-consumer -> cat-home ，Modules之间的设计使用了典型的模板模式。
cat-home的setup
在上一章讲到模块初始化的时候， 讲到setup安装cat-home模块，对于客户端的请求的监听处理，就是在这里完成的。
@Named(type = Module.class, value = CatHomeModule.ID)
public class CatHomeModule extends AbstractModule {
@Override
protected void setup(ModuleContext ctx) throws Exception {
if (!isInitialized()) {
File serverConfigFile = ctx.getAttribute("cat-server-config-file");
ServerConfigManager serverConfigManager = ctx.lookup(ServerConfigManager.class);
final TcpSocketReceiver messageReceiver = ctx.lookup(TcpSocketReceiver.class); serverConfigManager.initialize(serverConfigFile);
messageReceiver.init(); Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
messageReceiver.destory();
}
});
}
}
}
1、读取 server.xml 配置，装进配置管理器(ServerConfigManager)。
2、创建消息接收器 final TcpSocketReceiver messageReceiver;
3、messageReceiver.init() 初始化服务,采用的经典的 netty reactor 模型。
4、注册一个JVM关闭的钩子，在进程挂掉的时候，执行一些清理现场的代码。
TcpSocketReceiver--- netty reactor 模式的应用
我们来看看CatHomeModule对TcpSocketReceiver的初始化做了什么，如下源码：
public final class TcpSocketReceiver implements LogEnabled {
public void init() {
try {
startServer(m_port);
} catch (Throwable e) {
m_logger.error(e.getMessage(), e);
}
}
public synchronized void startServer(int port) throws InterruptedException {
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);
bootstrap.channel(linux ? EpollServerSocketChannel.class : NioServerSocketChannel.class);
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decode", new MessageDecoder());
}
});
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);
}
}
}
1、创建EventLoopGroup对象， EventLoopGroup是用来处理IO操作的多线程事件循环器，m_bossGroup作为一个acceptor负责接收来自客户端的请求，然后分发给m_workerGroup用来所有的事件event和channel的IO。
2、创建ServerBootstrap对象，ServerBootstrap 是一个启动Epoll(非Linux为NIO)服务的辅助启动类，他将设置bossGroup和workerGroup两个多线程时间循环器。
3、接下来的channel()方法设置了ServerBootstrap 的 ChannelFactory，这里传入的参数是EpollServerSocketChannel.class （非Linux为NioServerSocketChannel.class），也就是说这个ChannelFactory创建的就是EpollServerSocketChannel/NioServerSocketChannel的实例。
Channel是Netty的核心概念之一，它是Netty网络通信的主体，他从EventLoopGroup获得一个EventLoop，并注册到该EventLoop，channel生命周期内都和该EventLoop在一起，由它负责对网络通信连接的打开、关闭、连接和读写操作。如果是对于读写事件，执行线程调度pipeline来处理用户业务逻辑。
4、接下来bootstrap.childHandler的目的是添加一个handler，用来监听已经连接的客户端的Channel的动作和状态，传入的 ChannelInitializer重写了initChannel方法，这个方法在Channel被注册到EventLoop的时候会被调用。
5、initChannel会创建ChannelPipeline对象，并调用addLast添加ChannelHandler。有网络请求时，ChannelPipeline会调用ChannelHandler来处理，有ChannelInboundHandler和ChannelOutboundHandler两种，ChannelPipeline会从头到尾顺序调用ChannelInboundHandler处理网络请求内容，从尾到头调用ChannelOutboundHandler处理网络请求内容。这也是Netty用来灵活处理网络请求的机制之一，因为使用的时候可以用多个decoder和encoder进行组合，从而适应不同的网络协议。而且这种类似分层的方式可以让每一个Handler专注于处理自己的任务而不用管上下游，这也是pipeline机制的特点。这跟TCP/IP协议中的五层和七层的分层机制有异曲同工之妙。
在这里，ChannelPipeline添加的 ChannelHandler 是MessageDecoder ，MessageDecoder的祖先类实现了ChannelHandler接口，他本质上还是一个Handler，是网络IO事件具体处理类，当客户端将日志数据上传到服务器之后，会交给MessageDecoder 解码数据，然后进行后续处理。
6、调用 childOption 设置 channel 的参数。
7、最后调用bind()方法启动服务。
关于netty ，我就讲到这里，网上关于netty框架的文章非常多，大家可以自行去查。
消息的解码
上一章我们讲到Netty将接收到的消息交给 MessageDecoder 去做解码，解码是交由PlainTextMessageCodec对象将接收到的字节码反序列化为MessageTree对象(所有的消息都是由消息树来组织)，具体的解码逻辑在这里暂不做详细阐述，在第三章我们会阐述编码过程，解码只是编码的一个逆过程。
解码之后调用 DefaultMessageHandler 的 handle方法对消息进行处理，handle方法就干了一件事情，就是调用 m_consumer.consume(tree) 方法去消费消息树，在消费模块，CAT实现了队列化，异步化，在消息消费章节会详细阐述。
当然netty handler也是支持异步处理的，我们也可以将 DefaultMessageHandler 像 MessageDecoder那样向netty注册handler, 再由netty来做线程池分发。
public class MessageDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
if (buffer.readableBytes() < 4) {
return;
}
buffer.markReaderIndex();
int length = buffer.readInt();
...
ByteBuf readBytes = buffer.readBytes(length + 4);
...
DefaultMessageTree tree = (DefaultMessageTree) m_codec.decode(readBytes);
readBytes.resetReaderIndex();
tree.setBuffer(readBytes);
m_handler.handle(tree);
m_processCount++;
...
}
}