Vertx实战之如何追踪异步函数调用

背景

日常开发中我们经常需要处理各种系统问题,而这些系统问题通常由一些非预期的因素引起(比如非预期的输入,内存不够,网络波动等)。此时就需要知道

  1. 本次系统问题影响了谁?
  2. 如果本次系统问题是因为非预期的输入而导致的,那么这个非预期的输入是什么?

上述两点在同步编程里可以通过全局try-catch来实现。但在异步编程里该怎么办呢?


思路

我的想法是绑定一个唯一id到一次完整的请求调用中(这个完整取决于我们想要监控的范围),无论程序执行到何处,我们总能拿到与请求一一对应的调用上下文的id。

我将分两种场景进行讨论。一种是同步或基于jut下的线程池类进行异步调用的应用程序(如基于spring5之前的应用程序)。另外一种基于vertx(底层基于netty)的应用程序。


技术栈

vert.x 4netty 4alibaba TransmittableThreadLocaljboss javassist、hotspot类加载&方法调用


1. 同步or基于jdk的异步

关于这种场景,有以下几种实现:

  1. 程序中的函数调用是完全同步的

    则可通过java.lang.ThreadLocal绑定一个唯一id到当前线程。入口处生成,出口处销毁。
    如果我们用的是logback,也可通过 org.slf4j.MDC 来实现这个功能。其内部也是基于ThreadLocal。

  2. 程序中函数调用是异步的,异步通过新建java.lang.Thread 的方式实现。

    这里可通过java.lang.InheritableThreadLocal 绑定一个唯一id到当前线程。Thread的构造函数里会自动把通过InheritableThreadLocal 绑定到当前线程的数据拷贝到这个正在创建的Thread 里。所以,只要这个Thread 是在我们需要监控的区域(当前线程通过InheritableThreadLocal绑定了id后)创建的,就能实现这个唯一id的跨线程传递。

  3. 程序中函数调用是异步的,异步通过java.util.concurrent 下的线程池类实现。

    由于线程池中的线程是可以复用的,所以,如果我们往线程池中丢任务时,有两种情况:

    1. 线程池创建新的线程来执行该任务(比如线程池中的线程数<coreSize)。这种情况下InheritableThreadLocal 依然是有效的。
    2. 线程池把任务分配给了已有线程(比如线程池中的线程数≥coreSize并且待执行任务队列没有填满)。这种情况下InheritableThreadLocal 是无法生效的(线程池中的存活线程会循环拉取等待任务队列中的task去执行,而这个过程是没有InheritableThreadLocal 拷贝的)。所以,这里可以用阿里的[TransmittableThreadLocal](https://github.com/alibaba/transmittable-thread-local) 组件来实现这个唯一id在线程池中的传递(其核心原理是在Runnable 中用一个变量来存储当前线程的ThreadLocal值,当线程执行此Runnable时再拿出来)。

实现思路总结

异步环境下的关键在于如何跨线程传递ThreadLocal的值


2. Vertx 中的实现

vertx是一个类似于spring系列的用于构建分布式微服务的框架。基于vertx构建的应用程序最大的特点是无阻塞&全异步。

vertx里的异步主要分两种。一种是在eventbus上传递消息的异步。另一种是基于netty的io操作异步

1. eventbus上异步传递消息

先来看一个简单的示例程序

  static final String ADDRESS_1 = "address1";
  static final String MESSAGE_1 = "message1";
  static final String REPLY_MESSAGE_1 = "replyMessage1";

    public static void sendMsgByEventbus() {
        //初始化一个具有默认设置的vertx实例
        Vertx vertx = Vertx.vertx();

        //注册一个handler/consumer到eventbus
        vertx.eventBus().consumer(ADDRESS_1, event -> {
            log.info("receive msg:{}", event.body());
            event.reply(REPLY_MESSAGE_1);
        });

        //通过eventbus发送消息给刚注册的handler
        vertx.eventBus().request(ADDRESS_1, MESSAGE_1, reply -> {
            log.info("receive replied msg:{}", reply.result().body());
        });
    }

下面是程序执行的结果

[vert.x-eventloop-thread-0] - receive msg:message1
[vert.x-eventloop-thread-0] - receive replied msg:replyMessage1

可以看到发起调用的是主线程,处理调用跟处理回调的是线程vert.x-eventloop-thread-0

那么,这个异步是如何实现的呢?显然,异步最明显的体现就是最后发送消息这里。所以我们就从Eventbus#request 逐步深入。

大概流程是这样的:

  1. eventbus中根据我们传入的address拿到所有注册到这个地址上的handler。
  2. 将对应handler对传入消息的处理包装成一个runnable丢进一个queue
  3. eventLoop Thread从这个queue中抓取task执行

关键代码片段如下:

//与本主题无关的只展示调用链路供大家参考
//Eventbus#request
//  EventbusImpl#request
//    EventbusImpl#sendOrPubInternal
//      EventbusImpl#sendOrPubInternal
//        EventbusImpl#OutboundDeliveryContext
//          EventbusImpl#sendOrPub
//            EventbusImpl#sendLocally
//              EventbusImpl#deliverMessageLocally

protected ReplyException deliverMessageLocally(MessageImpl msg) {
      //1.找出跟msg.address()的handlerHolder
      ConcurrentCyclicSequence<HandlerHolder> handlers = handlerMap.get(msg.address());
      if (handlers != null) {
        for (HandlerHolder holder: handlers) {
          //2.依次调用这些handler的receive方法
          holder.handler.receive(msg.copyBeforeReceive());
        }
        return null;
      }
}

void receive(MessageImpl msg) {
      //3.匿名异步任务进队列
      context.nettyEventLoop().execute(() -> {
         doReceive(msg);
      });
}

//SingleThreadEventExecutor#execute
private void execute(Runnable task, boolean immediate) {
    //比较当前线程是否是eventLoop线程,是则返回true
    //由于我们在main里调用的eventbus#request,所以这里是false
    boolean inEventLoop = inEventLoop();

    //4. 添加到queue中,后面执行任务的线程会调用该queue的poll方法
    addTask(task);

    if (!inEventLoop) {
        //如果当前eventLoop对象的thread为空则创建一个Thread绑定到当前eventLoop
        startThread();
    }
}

//eventLoop中运行的thread通过这个方法从上面说的queue中拿task执行
//外层会循环执行该runAllTasks,直到Eventloop#shutdownGracefully被执行
//参考NioEventLoop#run
protected boolean runAllTasks(long timeoutNanos) {
      Runnable task = pollTask();
      
      for (;;) {
          //5.执行task
          safeExecute(task);

          runTasks ++;
                    
          //如果队列中还有task,则继续执行
          task = pollTask();
          if (task == null) {
              lastExecutionTime = ScheduledFutureTask.nanoTime();
              break;
          }
      }
      return true;
  }

protected static Runnable pollTaskFrom(Queue<Runnable> taskQueue) {
    for (;;) {
        Runnable task = taskQueue.poll();
        if (task != WAKEUP_TASK) {
            return task;
        }
    }
}

//safeExecute会去执行第3步中的匿名任务
protected boolean doReceive(Message<T> message) {
        Handler<Message<T>> theHandler = handler;
        
        deliver(theHandler, message);
        return true;
}

//MessageConsumerImpl#deliver
//  MessageConsumerImpl#dispatch
//    AbstractContext#dispatch
//      InboundDeliveryContext#dispatch   ++ 上一步的匿名Runnable会调用 
//        InboundDeliveryContext#next
//          MessageConsumerImpl#dispatch
//            DuplicatedContext#emit
//              EventLoopContext#emit
<T> void emit(AbstractContext ctx, T argument, Handler<T> task) {
    try {
      //6.执行handler/task中的事件处理
      //这个task就是我们之前通过Eventbus#consumer注册进来的
      task.handle(argument);
    } catch (Throwable t) {
      reportException(t);
    } finally {
      ctx.endDispatch(prev);
    }
}

我删减了一些不重要的部分,以便更易于理解。

通过上面的代码片段我们可以发现,第3步的SingleThreadEventExecutor#execute往队列里push任务似乎是解决问题(跨线程传递唯一id)的关键。所以,如果我们能够在这个task进入queue之前往这个task中塞上当前线程的ThreadLocal值,待到这个taskrun方法被执行时再把这个ThreadLocal值拿出来塞到当前线程(执行这个task的线程)的ThreadLocal,问题就解决了。

显然,我们需要修改这些类的代码。那么,怎么实现这个功能呢?

java.lang.instrument 提供了一系列运行时替换class字节码的技术。而jboss Javaassist 则提供了一些列修改字节码的接口。所以我们只需要结合这两项技术再通过javaagent指令即可达成我们的目的。

修改字节码的实现可参考这里


关于类加载

由于参与修改所有类都是由appClassLoader加载的,所以不会出现什么问题。但是,如果我们要按照上述设计对jdk自带(jre/lib)的一些类进行修改呢?比如说实现ThreadLocalThreadPoolExecutor#execute中的传递,思路似乎没多大区别,无非就是对传入ThreadPoolExecutor#execute中的Runnable做一层包装(设该包装类为类A)。然后当我们运行是却会发现每次执行execute方法时都会得到一个NoClassDefFoundException,无法找到类A

这是为什么呢?明明类A对应的class文件是存在的(build目录下),但虚拟机为什么在执行execute方法时找不到它呢?

这是因为ThreadPoolExecutor这个类是由BootStrapClassLoader加载的,而正常情况下类A并不处于BootStrapClassLoader的查找范围之中,所以就引发了这个异常。概览如下:

  • 在虚拟机第一次执行到execute方法中我们修改过的代码时,此时类A在此处仅仅是一个符号引用,这时必然会请求虚拟机去解析这个符号引用。这里会去根据加载ThreadPoolExecutor的类加载器(BootStrapClassLoader)以及类A的类名在内存中全局查找这个类,查找不到则会尝试加载这个类,如果这两个过程都找不到对应的类,就会抛出上述异常。

像alibaba的TransmittableThreadLocal 是需要修改rt.jar中的类的(比如ThreadPoolExecutor),这些核心类都是由BootstrapClassloader加载的。但是我们引进来的TransmittableThreadLocal 默认由AppClassLoader加载,这势必会导致在重新加载被修改的由class时抛出NoClassDefFoundException。所以TransmittableThreadLocal参与核心类库修改的类必须要被添加到bootClassPath中(由BootstrapClassLoader加载)。

2. 基于netty的异步io操作

vertx的io操作基于netty,netty的io多路复用基于java nio,nio只能说是非阻塞式io。但是netty提供给上层应用程序的io操作确实异步的。netty会不停的轮询就绪的io事件,然后把vertx感兴趣的事件包装好通知到vertx(比如有数据可读的时候),最后vertx再通知到我们的业务层。

其实这里的实现思路是一样的,就不赘述了。简单点就是在往vertx中塞回调时对这个回调进行上述包装即可。


总结

本文主要探讨了如何在基于vertx的异步环境中追踪一次完整的函数调用,以及实现过程中可能会碰到的问题。

上述内容的源码可在我的github上找到。

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 199,711评论 5 468
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 83,932评论 2 376
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 146,770评论 0 330
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 53,799评论 1 271
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 62,697评论 5 359
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,069评论 1 276
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,535评论 3 390
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,200评论 0 254
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,353评论 1 294
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,290评论 2 317
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,331评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,020评论 3 315
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,610评论 3 303
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,694评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,927评论 1 255
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,330评论 2 346
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 41,904评论 2 341

推荐阅读更多精彩内容

  • Java异步编程实战 chap1 认识异步编程 异步编程概念与作用在使用同步编程方式时,由于每个线程同时只能发起一...
    landon30阅读 1,175评论 0 0
  • Java异步编程实战 认识异步编程 异步编程概念与作用 同步编程简单,但是容易出现性能瓶颈;并行多个线程可以提高系...
    老干妈炒饭阅读 758评论 0 1
  • chap8 高性能异步编程框架和中间件 NettyNetty框架将网络编程逻辑与业务逻辑处理分离开来,其内部会自动...
    landon30阅读 778评论 0 0
  • 背景 新的工作新的开始,先描述下问题的背景,项目中为了解决多数据源聚合快速响应问题,启用线程池并发调用多数据源服务...
    Yang0503阅读 3,567评论 0 3
  • 我是黑夜里大雨纷飞的人啊 1 “又到一年六月,有人笑有人哭,有人欢乐有人忧愁,有人惊喜有人失落,有的觉得收获满满有...
    陌忘宇阅读 8,518评论 28 53