深入详解美团点评CAT跨语言服务监控(二) CAT服务端初始化

Cat模块

深入详解美团点评CAT跨语言服务监控(二) 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初始化

深入详解美团点评CAT跨语言服务监控(二) CAT服务端初始化

图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 认为是容器配置文件。
深入详解美团点评CAT跨语言服务监控(二) CAT服务端初始化

图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模块。

深入详解美团点评CAT跨语言服务监控(二) CAT服务端初始化

图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框架的文章非常多,大家可以自行去查。
深入详解美团点评CAT跨语言服务监控(二) CAT服务端初始化

消息的解码

深入详解美团点评CAT跨语言服务监控(二) CAT服务端初始化

  上一章我们讲到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++; ...
}
}
上一篇:centos7.6 ssh远程链接配置


下一篇:mysql数据库中利用select into outfile实现数据的备份与还原