arthas源码刨析：arthas 命令粗谈(3)

文章目录

dashboard
watch
retransform

前面介绍了 arthas 启动相关的代码并聊了聊怎么到一个 shellserver 的建立。
本篇我们来探讨一下几个使用频次非常高的命令是如何实现的。

dashboard

在这里插入图片描述
想看这个命令的主要原因是编程这些年来从来没有开发过 terminal 的这种比较花哨的界面，和 top 命令一样。

  public void process(final CommandProcess process) {

        Session session = process.session();
        timer = new Timer("Timer-for-arthas-dashboard-" + session.getSessionId(), true);

        // ctrl-C support
        process.interruptHandler(new DashboardInterruptHandler(process, timer));

        /*
         * 通过handle回调，在suspend和end时停止timer，resume时重启timer
         */
        Handler<Void> stopHandler = new Handler<Void>() {
            @Override
            public void handle(Void event) {
                stop();
            }
        };

        Handler<Void> restartHandler = new Handler<Void>() {
            @Override
            public void handle(Void event) {
                restart(process);
            }
        };
        process.suspendHandler(stopHandler);
        process.resumeHandler(restartHandler);
        process.endHandler(stopHandler);

        // q exit support
        process.stdinHandler(new QExitHandler(process));

        // start the timer
        timer.scheduleAtFixedRate(new DashboardTimerTask(process), 0, getInterval());
    }

获取会话
```
Session session = process.session();
```
- 调用process对象的session方法获取当前会话对象，并赋值给session变量。process
创建定时器
```
timer = new Timer("Timer-for-arthas-dashboard-" + session.getSessionId(), true);
```
- 创建一个定时器对象，命名为"Timer-for-arthas-dashboard-" + session.getSessionId()，其中session.getSessionId()获取会话ID。定时器的主要作用是进行界面刷新。
- 第二个参数true表示这个定时器是守护线程，当所有非守护线程结束时，程序会自动退出。
处理中断
```
process.interruptHandler(new DashboardInterruptHandler(process, timer));
```
- 设置一个中断处理器DashboardInterruptHandler，用于处理中断信号（如Ctrl+C）。

处理挂起、恢复和结束事件

Handler<Void> stopHandler = new Handler<Void>() {
    @Override
    public void handle(Void event) {
        stop();
    }
};

Handler<Void> restartHandler = new Handler<Void>() {
    @Override
    public void handle(Void event) {
        restart(process);
    }
};
process.suspendHandler(stopHandler);
process.resumeHandler(restartHandler);
process.endHandler(stopHandler);

定义了两个处理器stopHandler和restartHandler，分别用于处理挂起、恢复和结束事件。
stopHandler调用stop方法停止定时器。
restartHandler调用restart(process)方法重新启动定时器。
将这些处理器分别注册到process对象的suspendHandler、resumeHandler和endHandler中。

处理退出命令
```
process.stdinHandler(new QExitHandler(process));
```
- 设置一个标准输入处理器QExitHandler，用于处理退出命令（如输入q）。
启动定时器
```
timer.scheduleAtFixedRate(new DashboardTimerTask(process), 0, getInterval());
```
- 使用timer对象定期执行DashboardTimerTask任务。
- 周期性执行 DashboardTimerTask ，统计信息从 jvm 的 MXBean 中获取并组装 dashboardModel 再后面跟踪，就能发现界面绘制是属于 ProcessImpl 对命令结果的处理，发现没有，arthas 是一个很完整的解耦设计，职责分明。如何绘制界面在 view 文件夹，比如该命令

在这里插入图片描述
再往后看，界面就是做的以行为单元的字符串，输出到标准输出中，stdout。

watch

下面的 watch 命令在分析性能问题是出场率最高的命令，极大的释放了开发的背锅压力。
WatchCommand 继承了 EnhancerCommand ，下面的代码就是增强的核心代码。一般来说我们都知道使用 javaagent 可以在加载到jvm之前做增强，也就是 premain 方法，但 jvm也提供了 attach 的增强，即 agentmain，这也是在上一篇中提到过的

protected void enhance(CommandProcess process) {
        Session session = process.session();
        if (!session.tryLock()) {
            String msg = "someone else is enhancing classes, pls. wait.";
            process.appendResult(new EnhancerModel(null, false, msg));
            process.end(-1, msg);
            return;
        }
        EnhancerAffect effect = null;
        int lock = session.getLock();
        try {
            Instrumentation inst = session.getInstrumentation();
            AdviceListener listener = getAdviceListenerWithId(process);
            if (listener == null) {
                logger.error("advice listener is null");
                String msg = "advice listener is null, check arthas log";
                process.appendResult(new EnhancerModel(effect, false, msg));
                process.end(-1, msg);
                return;
            }
            boolean skipJDKTrace = false;
            if(listener instanceof AbstractTraceAdviceListener) {
                skipJDKTrace = ((AbstractTraceAdviceListener) listener).getCommand().isSkipJDKTrace();
            }

            Enhancer enhancer = new Enhancer(listener, listener instanceof InvokeTraceable, skipJDKTrace, getClassNameMatcher(), getClassNameExcludeMatcher(), getMethodNameMatcher());
            // 注册通知监听器
            process.register(listener, enhancer);
            effect = enhancer.enhance(inst, this.maxNumOfMatchedClass);

            if (effect.getThrowable() != null) {
                String msg = "error happens when enhancing class: "+effect.getThrowable().getMessage();
                process.appendResult(new EnhancerModel(effect, false, msg));
                process.end(1, msg + ", check arthas log: " + LogUtil.loggingFile());
                return;
            }

            if (effect.cCnt() == 0 || effect.mCnt() == 0) {
                // no class effected
                if (!StringUtils.isEmpty(effect.getOverLimitMsg())) {
                    process.appendResult(new EnhancerModel(effect, false));
                    process.end(-1);
                    return;
                }
                // might be method code too large
                process.appendResult(new EnhancerModel(effect, false, "No class or method is affected"));

                String smCommand = Ansi.ansi().fg(Ansi.Color.GREEN).a("sm CLASS_NAME METHOD_NAME").reset().toString();
                String optionsCommand = Ansi.ansi().fg(Ansi.Color.GREEN).a("options unsafe true").reset().toString();
                String javaPackage = Ansi.ansi().fg(Ansi.Color.GREEN).a("java.*").reset().toString();
                String resetCommand = Ansi.ansi().fg(Ansi.Color.GREEN).a("reset CLASS_NAME").reset().toString();
                String logStr = Ansi.ansi().fg(Ansi.Color.GREEN).a(LogUtil.loggingFile()).reset().toString();
                String issueStr = Ansi.ansi().fg(Ansi.Color.GREEN).a("https://github.com/alibaba/arthas/issues/47").reset().toString();
                String msg = "No class or method is affected, try:\n"
                        + "1. Execute `" + smCommand + "` to make sure the method you are tracing actually exists (it might be in your parent class).\n"
                        + "2. Execute `" + optionsCommand + "`, if you want to enhance the classes under the `" + javaPackage + "` package.\n"
                        + "3. Execute `" + resetCommand + "` and try again, your method body might be too large.\n"
                        + "4. Match the constructor, use `<init>`, for example: `watch demo.MathGame <init>`\n"
                        + "5. Check arthas log: " + logStr + "\n"
                        + "6. Visit " + issueStr + " for more details.";
                process.end(-1, msg);
                return;
            }

            // 这里做个补偿,如果在enhance期间,unLock被调用了,则补偿性放弃
            if (session.getLock() == lock) {
                if (process.isForeground()) {
                    process.echoTips(Constants.Q_OR_CTRL_C_ABORT_MSG + "\n");
                }
            }

            process.appendResult(new EnhancerModel(effect, true));

            //异步执行，在AdviceListener中结束
        } catch (Throwable e) {
            String msg = "error happens when enhancing class: "+e.getMessage();
            logger.error(msg, e);
            process.appendResult(new EnhancerModel(effect, false, msg));
            process.end(-1, msg);
        } finally {
            if (session.getLock() == lock) {
                // enhance结束后解锁
                process.session().unLock();
            }
        }
    }

这个方法的核心就在这几行代码中：

			Enhancer enhancer = new Enhancer(listener, listener instanceof InvokeTraceable, skipJDKTrace, getClassNameMatcher(), getClassNameExcludeMatcher(), getMethodNameMatcher());
            // 注册通知监听器
            process.register(listener, enhancer);
            effect = enhancer.enhance(inst, this.maxNumOfMatchedClass);

/**
     * 对象增强
     *
     * @param inst              inst
     * @param maxNumOfMatchedClass 匹配的class最大数量
     * @return 增强影响范围
     * @throws UnmodifiableClassException 增强失败
     */
    public synchronized EnhancerAffect enhance(final Instrumentation inst, int maxNumOfMatchedClass) throws UnmodifiableClassException {
        // 获取需要增强的类集合
        this.matchingClasses = GlobalOptions.isDisableSubClass
                ? SearchUtils.searchClass(inst, classNameMatcher)
                : SearchUtils.searchSubClass(inst, SearchUtils.searchClass(inst, classNameMatcher));

        if (matchingClasses.size() > maxNumOfMatchedClass) {
            affect.setOverLimitMsg("The number of matched classes is " +matchingClasses.size()+ ", greater than the limit value " + maxNumOfMatchedClass + ". Try to change the limit with option '-m <arg>'.");
            return affect;
        }
        // 过滤掉无法被增强的类
        List<Pair<Class<?>, String>> filtedList = filter(matchingClasses);
        if (!filtedList.isEmpty()) {
            for (Pair<Class<?>, String> filted : filtedList) {
                logger.info("ignore class: {}, reason: {}", filted.getFirst().getName(), filted.getSecond());
            }
        }

        logger.info("enhance matched classes: {}", matchingClasses);

        affect.setTransformer(this);

        try {
            ArthasBootstrap.getInstance().getTransformerManager().addTransformer(this, isTracing);

            // 批量增强
            if (GlobalOptions.isBatchReTransform) {
                final int size = matchingClasses.size();
                final Class<?>[] classArray = new Class<?>[size];
                arraycopy(matchingClasses.toArray(), 0, classArray, 0, size);
                if (classArray.length > 0) {
                    inst.retransformClasses(classArray);
                    logger.info("Success to batch transform classes: " + Arrays.toString(classArray));
                }
            } else {
                // for each 增强
                for (Class<?> clazz : matchingClasses) {
                    try {
                        inst.retransformClasses(clazz);
                        logger.info("Success to transform class: " + clazz);
                    } catch (Throwable t) {
                        logger.warn("retransform {} failed.", clazz, t);
                        if (t instanceof UnmodifiableClassException) {
                            throw (UnmodifiableClassException) t;
                        } else if (t instanceof RuntimeException) {
                            throw (RuntimeException) t;
                        } else {
                            throw new RuntimeException(t);
                        }
                    }
                }
            }
        } catch (Throwable e) {
            logger.error("Enhancer error, matchingClasses: {}", matchingClasses, e);
            affect.setThrowable(e);
        }

        return affect;
    }

增强的关于字节码增强部分在 WatchAdviceListener 中，相当于做了一个 AOP。而这些 AdviceListenerAdapter 的调用链： SpyImpl ---- SpyAPI — SpyInterceptors — Enhancer.transform

class WatchAdviceListener extends AdviceListenerAdapter {

    private static final Logger logger = LoggerFactory.getLogger(WatchAdviceListener.class);
    private final ThreadLocalWatch threadLocalWatch = new ThreadLocalWatch();
    private WatchCommand command;
    private CommandProcess process;

    public WatchAdviceListener(WatchCommand command, CommandProcess process, boolean verbose) {
        this.command = command;
        this.process = process;
        super.setVerbose(verbose);
    }

    private boolean isFinish() {
        return command.isFinish() || !command.isBefore() && !command.isException() && !command.isSuccess();
    }

    @Override
    public void before(ClassLoader loader, Class<?> clazz, ArthasMethod method, Object target, Object[] args)
            throws Throwable {
        // 开始计算本次方法调用耗时
        threadLocalWatch.start();
        if (command.isBefore()) {
            watching(Advice.newForBefore(loader, clazz, method, target, args));
        }
    }

    @Override
    public void afterReturning(ClassLoader loader, Class<?> clazz, ArthasMethod method, Object target, Object[] args,
                               Object returnObject) throws Throwable {
        Advice advice = Advice.newForAfterReturning(loader, clazz, method, target, args, returnObject);
        if (command.isSuccess()) {
            watching(advice);
        }

        finishing(advice);
    }

    @Override
    public void afterThrowing(ClassLoader loader, Class<?> clazz, ArthasMethod method, Object target, Object[] args,
                              Throwable throwable) {
        Advice advice = Advice.newForAfterThrowing(loader, clazz, method, target, args, throwable);
        if (command.isException()) {
            watching(advice);
        }

        finishing(advice);
    }

    private void finishing(Advice advice) {
        if (isFinish()) {
            watching(advice);
        }
    }


    private void watching(Advice advice) {
        try {
            // 本次调用的耗时
            double cost = threadLocalWatch.costInMillis();
            boolean conditionResult = isConditionMet(command.getConditionExpress(), advice, cost);
            if (this.isVerbose()) {
                process.write("Condition express: " + command.getConditionExpress() + " , result: " + conditionResult + "\n");
            }
            if (conditionResult) {
                // TODO: concurrency issues for process.write

                Object value = getExpressionResult(command.getExpress(), advice, cost);

                WatchModel model = new WatchModel();
                model.setTs(new Date());
                model.setCost(cost);
                model.setValue(new ObjectVO(value, command.getExpand()));
                model.setSizeLimit(command.getSizeLimit());
                model.setClassName(advice.getClazz().getName());
                model.setMethodName(advice.getMethod().getName());
                if (advice.isBefore()) {
                    model.setAccessPoint(AccessPoint.ACCESS_BEFORE.getKey());
                } else if (advice.isAfterReturning()) {
                    model.setAccessPoint(AccessPoint.ACCESS_AFTER_RETUNING.getKey());
                } else if (advice.isAfterThrowing()) {
                    model.setAccessPoint(AccessPoint.ACCESS_AFTER_THROWING.getKey());
                }

                process.appendResult(model);
                process.times().incrementAndGet();
                if (isLimitExceeded(command.getNumberOfLimit(), process.times().get())) {
                    abortProcess(process, command.getNumberOfLimit());
                }
            }
        } catch (Throwable e) {
            logger.warn("watch failed.", e);
            process.end(-1, "watch failed, condition is: " + command.getConditionExpress() + ", express is: "
                    + command.getExpress() + ", " + e.getMessage() + ", visit " + LogUtil.loggingFile()
                    + " for more details.");
        }
    }
}

Enhance 是个自定义的 ClassTransformer ，在 Instrumentation 进行 addTransformer 时，将 enhance 增强纳进管理，在 redefined 或者 retransformed 的时候调用 transform 方法。。
Instrumentation 的方法：
在这里插入图片描述

retransform

retransform 这个命令的牛逼之处是可以进行灵活的热加载，用过 jrebel 的同学应该比较清除，在开发时能使用热加载，compile 一下修改的功能就能生效在开发时能节省大量的时间。
在arthas 中 jad–mc–retransform 工具链实现热加载也是一个在线上快速修复的一个方法。

在这里插入图片描述