基于Cat的分布式调用追踪

Cat是美团点评出的一款APM工具，同类的产品也有不少，知名的开源产品如zipkin和pinpoint；国内收费的产品如oneapm。考虑到Cat在互联网公司的应用比较广，因此被纳入选型队列，我也有幸参与技术预言。

使用Cat断断续续将近两周的时间，感觉它还算是很轻量级的。文档相对来说薄弱一些，没有太全面的官方文档（官方文档大多是介绍每个名词是什么意思，界面是什么意思，部署方面比较欠缺）；但是好在有一个非常活跃的群，群里有很多经验丰富的高手，不会的问题基本都能得到解答。

下面就开始步入正题吧，本篇主要讲述一下如何利用Cat进行分布式的调用链追踪。

分布式开发基础

在最开始网站基本都是单节点的，由于业务逐渐发展，使用者开始增多，单节点已经无法支撑了。于是开始切分系统，把系统拆分成几个独立的模块，模块之间采用远程调用的方式进行通信。

那么远程调用是如何做到的呢？下面就用最古老的RMI的方式来举个例子吧！

RMI（Remote method invocation）是java从1.1就开始支持的功能，它支持跨进程间的方法调用。

大体上的原理可以理解为，服务端会持续监听一个端口。客户端通过proxy代理的方式远程调用服务端。即客户端会把方法的参数以字符串的的方式序列化传给服务端。服务端反序列化后调用本地的方法执行，执行结果再序列化返回给客户端。

服务端的代码可以参考如下：

  1.  
     
  2.  
    interface IBusiness extends Remote{
  3.  
    String echo(String message) throws RemoteException;
  4.  
    }
  5.  
    class BusinessImpl extends UnicastRemoteObject implements IBusiness {
  6.  
    public BusinessImpl() throws RemoteException {}
  7.  
    @Override
  8.  
    public String echo(String message) throws RemoteException {
  9.  
    return "hello,"+message;
  10.  
    }
  11.  
    }
  12.  
    public class RpcServer {
  13.  
    public static void main(String[] args) throws RemoteException, AlreadyBoundException, MalformedURLException {
  14.  
    IBusiness business = new BusinessImpl();
  15.  
    LocateRegistry.createRegistry(8888);
  16.  
    Naming.bind("rmi://localhost:8888/Business",business);
  17.  
    System.out.println("Hello, RMI Server!");
  18.  
    }
  19.  
    }

客户端的代码如下：

  1.  
    IBusiness business = (IBusiness) Naming.lookup("rmi://localhost:8888/Business");
  2.  
    business.echo("xingoo",ctx);

上面的例子就可以实现客户端跨进程调用的例子。

Cat监控

Cat的监控跟传统的APM产品差不多，模式都是相似的，需要一个agent在客户端进行埋点，然后把数据发送给服务端，服务端进行解析并存储。只要你埋点足够全，那么它是可以进行全面监控的。监控到的数据会首先按照某种规则进行消息的合并，合并成一个MessageTree，这个MessageTree会被放入BlockingQueue里面，这样就解决了多线程数据存储的问题。

队列会限制存储的MessageTree的个数，但是如果服务端挂掉，客户端也有可能因为堆积大量的心跳而导致内存溢出（心跳是Cat客户端自动向服务端发出的，里面包含了jvm本地磁盘IO等很多的内容，所以MesssageTree挺大的）。

因此数据在客户端的流程可以理解为：

TrasactionEvent-->MessageTree-->BlockingQueue-->netty发出网络流

即Transaction、Event等消息会先合并为消息树，以消息树为单位存储在内存中（并未进行本地持久化），专门有一个TcpSocketSender负责向外发送数据。

再说说服务端，服务端暂时看的不深，大体上可以理解为专门有一个TcpSocketReciever接收数据，由于数据在传输过程中是需要序列化的。因此接收后首先要进行decode，生成消息树。然后把消息放入BlockingQueue，有分析器不断的来队列拿消息树进行分析，分析后按照一定的规则把报表存储到数据库，把原始数据存储到本地文件中(默认是存储到本地)。

因此数据在服务端的流程大致可以理解为：

  1.  
    网络流-->decode反序列化-->BlockingQueue-->analyzer分析--->报表存储在DB
  2.  
    |---->原始数据存储在本地或hdfs

简单的Transaction例子

在Cat里面，消息大致可以分为几个类型：

  • Transaction 有可能出错、需要记录处理的时间的监控，比如SQL查询、URL访问等
  • Event 普通的监控，没有处理时间的要求，比如一次偶然的异常，一些基本的信息
  • Hearbeat 心跳检测，常常用于一些基本的指标监控，一般是一分钟一次
  • Metric 指标，比如有一个值，每次访问都要加一，就可以使用它

Transaction支持嵌套，即可以作为消息树的根节点，也可以作为叶子节点。但是Event、Heartbeat和Metric只能作为叶子节点。有了这种树形结构，就可以描述出下面这种调用链的结果了：

Transaction和Event的使用很简单，比如：

  1.  
    @RequestMapping("t")
  2.  
    public @ResponseBody String test() {
  3.  
    Transaction t = Cat.newTransaction("MY-TRANSACTION","test in TransactionTest");
  4.  
    try{
  5.  
    Cat.logEvent("EVENT-TYPE-1","EVENT-NAME-1");
  6.  
     
  7.  
    // ....
  8.  
     
  9.  
    }catch(Exception e){
  10.  
    Cat.logError(e);
  11.  
    t.setStatus(e);
  12.  
    }finally {
  13.  
    t.setStatus(Transaction.SUCCESS);
  14.  
    t.complete();
  15.  
    }
  16.  
    return "trasaction test!";
  17.  
    }

这是一个最基本的Transaction的例子。

分布式调用链监控

在分布式环境中，应用是运行在独立的进程中的，有可能是不同的机器，或者不同的服务器进程。那么他们如果想要彼此联系在一起，形成一个调用链，就需要通过几个ID进行串联。这种串联的模式，基本上都是一样的。

举个例子，A系统在aaa()中调用了B系统的bbb()方法，如果我们在aaa方法中埋点记录上面例子中的信息，在bbb中也记录信息，但是这两个信息是彼此独立的。因此就需要使用一个全局的id，证明他们是一个调用链中的调用方法。除此之外，还需要一个标识谁在调用它的ID，以及一个标识它调用的方法的ID。

总结来说，每个Transaction需要三个ID：

  • RootId，用于标识唯一的一个调用链
  • ParentId，父Id是谁？谁在调用我
  • ChildId，我在调用谁?

其实ParentId和ChildId有点冗余，但是Cat里面还是都加上吧！

那么问题来了，如何传递这些ID呢？在Cat中需要你自己实现一个Context，因为Cat里面只提供了一个内部的接口：

  1.  
    public interface Context {
  2.  
    String ROOT = "_catRootMessageId";
  3.  
    String PARENT = "_catParentMessageId";
  4.  
    String CHILD = "_catChildMessageId";
  5.  
     
  6.  
    void addProperty(String var1, String var2);
  7.  
     
  8.  
    String getProperty(String var1);
  9.  
    }

我们需要自己实现这个接口，并存储相关的ID：

  1.  
    public class MyContext implements Cat.Context,Serializable{
  2.  
     
  3.  
    private static final long serialVersionUID = 7426007315111778513L;
  4.  
     
  5.  
    private Map<String,String> properties = new HashMap<String,String>();
  6.  
     
  7.  
    @Override
  8.  
    public void addProperty(String s, String s1) {
  9.  
    properties.put(s,s1);
  10.  
    }
  11.  
     
  12.  
    @Override
  13.  
    public String getProperty(String s) {
  14.  
    return properties.get(s);
  15.  
    }
  16.  
    }

由于这个Context需要跨进程网络传输，因此需要实现序列化接口。

在Cat中其实已经给我们实现了两个方法logRemoteCallClient以及logRemoteCallServer，可以简化处理逻辑,有兴趣可以看一下Cat中的逻辑实现：

  1.  
    //客户端需要创建一个Context，然后初始化三个ID
  2.  
    public static void logRemoteCallClient(Cat.Context ctx) {
  3.  
    MessageTree tree = getManager().getThreadLocalMessageTree();
  4.  
    String messageId = tree.getMessageId();//获取当前的MessageId
  5.  
    if(messageId == null) {
  6.  
    messageId = createMessageId();
  7.  
    tree.setMessageId(messageId);
  8.  
    }
  9.  
     
  10.  
    String childId = createMessageId();//创建子MessageId
  11.  
    logEvent("RemoteCall", "", "0", childId);
  12.  
    String root = tree.getRootMessageId();//获取全局唯一的MessageId
  13.  
    if(root == null) {
  14.  
    root = messageId;
  15.  
    }
  16.  
     
  17.  
    ctx.addProperty("_catRootMessageId", root);
  18.  
    ctx.addProperty("_catParentMessageId", messageId);//把自己的ID作为ParentId传给调用的方法
  19.  
    ctx.addProperty("_catChildMessageId", childId);
  20.  
    }
  21.  
     
  22.  
    //服务端需要接受这个context，然后设置到自己的Transaction中
  23.  
    public static void logRemoteCallServer(Cat.Context ctx) {
  24.  
    MessageTree tree = getManager().getThreadLocalMessageTree();
  25.  
    String messageId = ctx.getProperty("_catChildMessageId");
  26.  
    String rootId = ctx.getProperty("_catRootMessageId");
  27.  
    String parentId = ctx.getProperty("_catParentMessageId");
  28.  
    if(messageId != null) {
  29.  
    tree.setMessageId(messageId);//把传过来的子ID作为自己的ID
  30.  
    }
  31.  
     
  32.  
    if(parentId != null) {
  33.  
    tree.setParentMessageId(parentId);//把传过来的parentId作为
  34.  
    }
  35.  
     
  36.  
    if(rootId != null) {
  37.  
    tree.setRootMessageId(rootId);//把传过来的RootId设置成自己的RootId
  38.  
    }
  39.  
     
  40.  
    }

这样，结合前面的RMI调用，整个思路就清晰多了.

客户端调用者的埋点：

  1.  
    @RequestMapping("t2")
  2.  
    public @ResponseBody String test2() {
  3.  
    Transaction t = Cat.newTransaction("Call","test2");
  4.  
    try{
  5.  
    Cat.logEvent("Call.server","localhost");
  6.  
    Cat.logEvent("Call.app","business");
  7.  
    Cat.logEvent("Call.port","8888");
  8.  
     
  9.  
    MyContext ctx = new MyContext();
  10.  
    Cat.logRemoteCallClient(ctx);
  11.  
     
  12.  
    IBusiness business = (IBusiness) Naming.lookup("rmi://localhost:8888/Business");
  13.  
    business.echo("xingoo",ctx);
  14.  
    }catch(Exception e){
  15.  
    Cat.logError(e);
  16.  
    t.setStatus(e);
  17.  
    }finally {
  18.  
    t.setStatus(Transaction.SUCCESS);
  19.  
    t.complete();
  20.  
    }
  21.  
    return "cross!";
  22.  
    }

远程被调用者的埋点：

  1.  
    interface IBusiness extends Remote{
  2.  
    String echo(String message,MyContext ctx) throws RemoteException;
  3.  
    }
  4.  
    class BusinessImpl extends UnicastRemoteObject implements IBusiness {
  5.  
    public BusinessImpl() throws RemoteException {}
  6.  
    @Override
  7.  
    public String echo(String message,MyContext ctx) throws RemoteException {
  8.  
    Transaction t = Cat.newTransaction("Service","echo");
  9.  
    try{
  10.  
    Cat.logEvent("Service.client","localhost");
  11.  
    Cat.logEvent("Service.app","cat-client");
  12.  
    Cat.logRemoteCallServer(ctx);
  13.  
    System.out.println(message);
  14.  
    }catch(Exception e){
  15.  
    Cat.logError(e);
  16.  
    t.setStatus(e);
  17.  
    }finally {
  18.  
    t.setStatus(Transaction.SUCCESS);
  19.  
    t.complete();
  20.  
    }
  21.  
    return "hello,"+message;
  22.  
    }
  23.  
    }
  24.  
    public class RpcServer {
  25.  
    public static void main(String[] args) throws RemoteException, AlreadyBoundException, MalformedURLException {
  26.  
    IBusiness business = new BusinessImpl();
  27.  
    LocateRegistry.createRegistry(8888);
  28.  
    Naming.bind("rmi://localhost:8888/Business",business);
  29.  
    System.out.println("Hello, RMI Server!");
  30.  
    }
  31.  
    }

需要注意的是，Service的client和app需要和Call的server以及app对应上，要不然图表是分析不出东西的！

最后

Cat对于一些分布式的开源框架，都有很好的集成，比如dubbo，有兴趣的可以查看它在script中的文档，结合上面的例子可以更好地理解。

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/duanxz/p/9326466.html