今天我们从使用和实现两个方面来聊一聊大众点评的Java应用实时监控系统–CAT,它目前已成为一个开源项目,见Github

目录

CAT能做些什么?

在此之前,先来想一想对于线上应用我们希望能监控些什么?可能有如下这些:

  • 机器状态信息。CPU负载、内存信息、磁盘使用率这些是必需的,另外可能还希望收集Java进程的数据,例如线程栈、堆、垃圾回收等信息,以帮助出现问题时快速debug。
  • 请求访问情况。例如请求个数、响应时间、处理状态,如果有处理过程中的时间分析那就更完美了。
  • 异常情况。譬如缓存服务时不时出现无响应,我们希望能够监控到这种异常,从而做进一步的处理。
  • 业务情况。例如订单量统计,销售额等等。

以上这些CAT都能支持。根据其官方文档,CAT支持如下5种监控消息:

  • Transaction。记录跨越系统边界的程序访问行为,比如远程调用,数据库调用,也适合执行时间较长的业务逻辑监控。
  • Event。用来记录一件事发生的次数,比如记录系统异常,它和transaction相比缺少了时间的统计,开销比transaction要小。
  • Heartbeat。表示程序内定期产生的统计信息, 如CPU%, MEM%, 连接池状态, 系统负载等。
  • Metric。用于记录业务指标、指标可能包含对一个指标记录次数、记录平均值、记录总和,业务指标最低统计粒度为1分钟。
  • Trace。用于记录基本的trace信息,类似于log4j的info信息,这些信息仅用于查看一些相关信息

在一个请求处理中可能产生有多种消息,CAT将其组织成消息树的形式。在处理开始时,默认开始一个类型为URL的Transaction,在这个Transaction中业务本身可以产生子消息。例如,产生一个数据库访问的子Transaction或者一个订单统计的Metric。结构如下所示:

message-tree

CAT的使用比较简单,接口也比较清晰,关于其使用请参考官方文档,这里不再赘述。本文主要讨论其客户端的设计与实现。

CAT客户端的设计

作为一个日志上报的通用客户端,考虑点至少有如下这些:

  • 为了尽可能减少对业务的影响,需要对消息进行异步处理。即业务线程将消息交给CAT客户端与CAT客户端上报这两个过程需要异步。
  • 为了达到实时的目的以及适应高并发的情况,客户端上报应该基于TCP而非HTTP开发。
  • 在线程安全的前提下尽可能的资源低消耗以及低延时。我们知道,线程竞争的情况是由于资源共享造成的,要达到线程安全通常需要减少资源共享或者加锁,而这两点则会导致系统资源冗余和高延时。

CAT客户端实现并不复杂,但这些点都考虑到了。它的架构如下所示:

cat-architecture

大概步骤为:

  1. 业务线程产生消息,交给消息Producer,消息Producer将消息存放在该业务线程消息栈中;
  2. 业务线程通知消息Producer消息结束时,消息Producer根据其消息栈产生消息树放置在同步消息队列中;
  3. 消息上报线程监听消息队列,根据消息树产生最终的消息报文上报CAT服务端。

下面我们来一步一步分析其源码。

CAT客户端的实现

CAT客户端实现在源码目录cat-client下,而cat-client的主要实现则依赖于它的com.dianping.cat.message包。该包结构如下:

category

com.dianping.cat.message中主要包含了internal、io、spi这三个目录:

  • internal目录包含主要的CAT客户端内部实现类;
  • io目录包含建立服务端连接、重连、消息队列监听、上报等io实现类;
  • spi目录为上报消息工具包,包含消息二进制编解码、转义等实现类。

其uml图如下所示(可以放大看):

uml

类的功能如下:

  • Message为所有上报消息的抽象,它的子类实现有Transaction、Metric、Event、HeartBeat、Trace这五种。
  • MessageProducer封装了所有接口,业务在使用CAT时只需要通过MessageProducer来操作。
  • MessageManager为CAT客户端核心类,相当于MVC中的Controller。
  • Context类保存消息上下文。
  • TransportManager提供发送消息的sender,具体实现有DefaultTransportManager,调用其getSender接口返回一个TcpSocketSender。
  • TcpSocketSender类负责发送消息。

Message

上面说到,Message有五类,分别为Transaction、Metric、Event、HeartBeat、Trace。其中Metric、Event、HeartBeat、Trace基本相同,保存的数据都为一个字符串;而Transaction则保存一个Message列表。换句话说,Transaction的结构为一个递归包含的结构,其他结构则为原子性结构。

下面为DefaultTransaction的关键数据成员及操作:

public class DefaultTransaction extends AbstractMessage implements Transaction {
private List<Message> m_children;
private MessageManager m_manager;
...

//添加子消息
public DefaultTransaction addChild(Message message) {
...
}

//Transaction结束时调用此方法
public void complete() {
...
m_manager.end(this); //调用MessageManager来结束Transaction
...
}

值得一提的是,Transaction(或者其他的Message)在创建时自动开始,消息结束时需要业务方调用complete方法,而在complete方法内部则调用MessageManager来完成消息。

MessageProducer

MessageProducer对业务方封装了CAT内部的所有细节,它的主要方法如下:

public void logError(String message, Throwable cause);
public void logEvent(String type, String name, String status, String nameValuePairs);
public void logHeartbeat(String type, String name, String status, String nameValuePairs);
public void logMetric(String name, String status, String nameValuePairs);
public void logTrace(String type, String name, String status, String nameValuePairs);
...
public Event newEvent(String type, String name);
public Event newEvent(Transaction parent, String type, String name);
public Heartbeat newHeartbeat(String type, String name);
public Metric newMetric(String type, String name);
public Transaction newTransaction(String type, String name);
public Trace newTrace(String type, String name);
...

logXXX方法为方法糖(造词小能手呵呵),这些方法在调用时需要传入消息数据,方法结束后消息自动结束。

newXXX方法返回相应的Message,业务方需要调用Message方法设置数据,并最终调用Message.complete()方法结束消息。

MessageProducer只是接口封装,消息处理主要实现依赖于MessageManager这个类。

MessageManager

MessageManager为CAT的核心类,但它只是定义了接口,具体实现为DefaultMessageManager。DefaultMessageManager这个类里面主要包含了两个功能类,ContextTransportManager,分别用于保存上下文和消息传输。TransportManager运行期间为单例对象,而Context则包装成ThreadLocal为每个线程保存上下文。

我们通过接口来了解DefaultMessageManager的主要功能:

public void add(Message message);
public void start(Transaction transaction, boolean forked);
public void end(Transaction transaction);

public void flush(MessageTree tree);

add()方法用来添加原子性的Message,也就是Metric、Event、HeartBeat、Trace。

start()和end()方法用来开始和结束Transaction这种消息。

flush()方法用来将当前业务线程的所有消息刷新到CAT服务端,当然,是异步的。

Context

Context用来保存消息上下文,我们可以通过它的主要接口来了解它功能:

public void add(Message message) {
if (m_stack.isEmpty()) {
MessageTree tree = m_tree.copy();

tree.setMessage(message);
flush(tree);
} else {
Transaction parent = m_stack.peek();

addTransactionChild(message, parent);
}
}

add方法主要添加原子性消息,它先判断该消息是否有上文消息(即判断是否处于一个Transaction中)。如果有则m_stack不为空并且将该消息添加到上文Transaction的子消息队列中;否则直接调用flush来将此原子性消息刷新到服务端。

public void start(Transaction transaction, boolean forked) {
if (!m_stack.isEmpty()) {
...
Transaction parent = m_stack.peek();
addTransactionChild(transaction, parent);
} else {
m_tree.setMessage(transaction);
}

if (!forked) {
m_stack.push(transaction);
}
}

start方法用来开始Transaction(Transaction是消息里比较特殊的一种),如果当前消息栈为空则证明该Transaction为第一个Transaction,使用消息树保存该消息,同时将该消息压栈;否则将当前Transaction保存到上文Transaction的子消息队列中,同时将该消息压栈。

public boolean end(DefaultMessageManager manager, Transaction transaction) {
if (!m_stack.isEmpty()) {
Transaction current = m_stack.pop();
...
if (m_stack.isEmpty()) {
MessageTree tree = m_tree.copy();

m_tree.setMessageId(null);
m_tree.setMessage(null);
...
manager.flush(tree); //刷新消息到CAT服务端
return true;
}
}

return false;
}

end方法用来结束Transaction,每次调用都会pop消息栈,如果栈为空则调用flush来刷新消息到CAT服务端。

综上,Context的m_stack的结构如下:

message-stack

Transaction之间是有引用的,因此在end方法中只需要将第一个Transaction(封装在MessageTree中)通过MessageManager来flush,在拼接消息时可以根据这个引用关系来找到所有的Transaction :)。

TransportManager和MessageSender

这两个类用来发送消息到服务端。MessageManager通过TransportManager获取到MessageSender,调用sender.send()方法来发送消息。 TransportManager和MessageSender关系如下:

transport

TCPSocketSender为MessageSender的具体子类,它里面主要的数据成员为:

private MessageCodec m_codec;
private MessageQueue m_queue = new DefaultMessageQueue(SIZE);
private ChannelManager m_manager;
  • MessageCodec:CAT基于TCP传输消息,因此在发送消息时需要对字符消息编码成字节流,这个编码的工作由MessageCodec负责实现。

  • MessageQueue:还记得刚才说业务方在添加消息时,CAT异步发送到服务端吗?在添加消息时,消息会被放置在TCPSocketSender的m_queue中,如果超出queue大小则抛弃消息。

  • ChannelManager:CAT底层使用netty来实现TCP消息传输,ChannelManager负责维护通信Channel。通俗的说,维护连接。

TCPSocketSender主要方法为initialize、send和run,分别介绍如下:

public void initialize() {
m_manager = new ChannelManager(m_logger, m_serverAddresses, m_queue, m_configManager, m_factory);

Threads.forGroup("cat").start(this);
Threads.forGroup("cat").start(m_manager);
...
}

initialize方法为初始化方法,在执行时主要创建两个线程,一个用来运行自身run方法(TCPSocketSender实现了Runnable接口)监听消息队列;另一个则用来执行ChannelManager维护通信Channel。

public void send(MessageTree tree) {
if (isAtomicMessage(tree)) {
boolean result = m_atomicTrees.offer(tree, m_manager.getSample());

if (!result) {
logQueueFullInfo(tree);
}
} else {
boolean result = m_queue.offer(tree, m_manager.getSample());

if (!result) {
logQueueFullInfo(tree);
}
}
}

send方法被MessageManager调用,把消息放置在消息队列中。

public void run() {
m_active = true;

while (m_active) {
ChannelFuture channel = m_manager.channel();

if (channel != null && checkWritable(channel)) {
try {
MessageTree tree = m_queue.poll();

if (tree != null) {
sendInternal(tree);
tree.setMessage(null);
}

} catch (Throwable t) {
m_logger.error("Error when sending message over TCP socket!", t);
}
} else {
try {
Thread.sleep(5);
} catch (Exception e) {
// ignore it
m_active = false;
}
}
}
}

private void sendInternal(MessageTree tree) {
ChannelFuture future = m_manager.channel();
ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(10 * 1024); // 10K

m_codec.encode(tree, buf);

int size = buf.readableBytes();
Channel channel = future.channel();

channel.writeAndFlush(buf);
if (m_statistics != null) {
m_statistics.onBytes(size);
}
}

run方法会一直执行直到进程退出,在循环里先获取通信Channel,然后发送消息。值得注意的是,sendInternal方法在执行时调用m_codec.encode(tree, buf),参数为消息树缓冲区。消息树里面其实只保存了一个消息,还记得刚才说的Transaction上下文引用吗?m_codec在encode的时候会判断消息类型是否为Transaction,如果为Transaction则会递归获取子Transaction,否则直接将该消息编码。具体实现可以参考源代码的PlainTextMessageCodec类的encode方法,此处不再赘述。

最后

本文主要分享了大众点评的Java应用监控系统–CAT的客户端实现,如有错漏恳请指正。以上。