聊聊skywaking的CommandService

本文主要研究一下skywalking的CommandService

CommandService

skywalking-6.6.0/apm-sniffer/apm-agent-core/old-src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandService.java

@DefaultImplementor
public class CommandService implements BootService, Runnable {

    private static final ILog LOGGER = LogManager.getLogger(CommandService.class);

    private volatile boolean isRunning = true;
    private ExecutorService executorService = Executors.newSingleThreadExecutor();
    private LinkedBlockingQueue<BaseCommand> commands = new LinkedBlockingQueue<BaseCommand>(64);
    private CommandSerialNumberCache serialNumberCache = new CommandSerialNumberCache();

    @Override
    public void prepare() throws Throwable {
    }

    @Override
    public void boot() throws Throwable {
        executorService.submit(new RunnableWithExceptionProtection(this, new RunnableWithExceptionProtection.CallbackWhenException() {
            @Override
            public void handle(final Throwable t) {
                LOGGER.error(t, "CommandService failed to execute commands");
            }
        }));
    }

    @Override
    public void run() {
        final CommandExecutorService commandExecutorService = ServiceManager.INSTANCE.findService(CommandExecutorService.class);

        while (isRunning) {
            try {
                BaseCommand command = commands.take();

                if (isCommandExecuted(command)) {
                    continue;
                }

                commandExecutorService.execute(command);
                serialNumberCache.add(command.getSerialNumber());
            } catch (InterruptedException e) {
                LOGGER.error(e, "Failed to take commands.");
            } catch (CommandExecutionException e) {
                LOGGER.error(e, "Failed to execute command[{}].", e.command().getCommand());
            } catch (Throwable e) {
                LOGGER.error(e, "There is unexpected exception");
            }
        }
    }

    private boolean isCommandExecuted(BaseCommand command) {
        return serialNumberCache.contain(command.getSerialNumber());
    }

    @Override
    public void onComplete() throws Throwable {

    }

    @Override
    public void shutdown() throws Throwable {
        isRunning = false;
        commands.drainTo(new ArrayList<BaseCommand>());
        executorService.shutdown();
    }

    public void receiveCommand(Commands commands) {
        for (Command command : commands.getCommandsList()) {
            try {
                BaseCommand baseCommand = CommandDeserializer.deserialize(command);

                if (isCommandExecuted(baseCommand)) {
                    LOGGER.warn("Command[{}] is executed, ignored", baseCommand.getCommand());
                    continue;
                }

                boolean success = this.commands.offer(baseCommand);

                if (!success && LOGGER.isWarnEnable()) {
                    LOGGER.warn("Command[{}, {}] cannot add to command list. because the command list is full.",
                        baseCommand.getCommand(), baseCommand.getSerialNumber());
                }
            } catch (UnsupportedCommandException e) {
                if (LOGGER.isWarnEnable()) {
                    LOGGER.warn("Received unsupported command[{}].", e.getCommand().getCommand());
                }
            }
        }
    }
}
  • CommandService实现了BootService接口及Runnable接口,其boot方法会往executorService注册自己的run方法;其shutdown方法会执行commands.drainTo(new ArrayList<BaseCommand>())及executorService.shutdown();run方法会循环从commands取出command,然后通过serialNumberCache判断是否已经执行,已经执行则跳过,否则通过commandExecutorService.execute(command)执行并添加到serialNumberCache中;它还提供了receiveCommand方法,该方法会通过commands.offer(baseCommand)将command添加到commands队列

CommandSerialNumberCache

skywalking-6.6.0/apm-sniffer/apm-agent-core/old-src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandSerialNumberCache.java

public class CommandSerialNumberCache {
    private static final int DEFAULT_MAX_CAPACITY = 64;
    private final Deque&lt;String&gt; queue;
    private final int maxCapacity;

    public CommandSerialNumberCache() {
        this(DEFAULT_MAX_CAPACITY);
    }

    public CommandSerialNumberCache(int maxCapacity) {
        queue = new LinkedBlockingDeque&lt;String&gt;(maxCapacity);
        this.maxCapacity = maxCapacity;
    }

    public void add(String number) {
        if (queue.size() &gt;= maxCapacity) {
            queue.pollFirst();
        }

        queue.add(number);
    }

    public boolean contain(String command) {
        return queue.contains(command);
    }
}
  • CommandSerialNumberCache使用Deque来做缓存,默认大小为DEFAULT_MAX_CAPACITY,add方法在queue的size大于等于maxCapacity时,会先queue.pollFirst(),然后再执行queue.add(number);其contain方法则是通过queue.contains(command)来判断

CommandExecutor

skywalking-6.6.0/apm-sniffer/apm-agent-core/old-src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandExecutor.java

public interface CommandExecutor {
    /**
     * @param command the command that is to be executed
     * @throws CommandExecutionException when the executor failed to execute the command
     */
    void execute(BaseCommand command) throws CommandExecutionException;
}
  • CommandExecutor接口定义了execute方法

CommandExecutorService

skywalking-6.6.0/apm-sniffer/apm-agent-core/old-src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandExecutorService.java

@DefaultImplementor
public class CommandExecutorService implements BootService, CommandExecutor {
    private Map&lt;String, CommandExecutor&gt; commandExecutorMap;

    @Override
    public void prepare() throws Throwable {
        commandExecutorMap = new HashMap&lt;String, CommandExecutor&gt;();

        // Register all the supported commands with their executors here
        commandExecutorMap.put(ServiceResetCommand.NAME, new ServiceResetCommandExecutor());
    }

    @Override
    public void boot() throws Throwable {

    }

    @Override
    public void onComplete() throws Throwable {

    }

    @Override
    public void shutdown() throws Throwable {

    }

    @Override
    public void execute(final BaseCommand command) throws CommandExecutionException {
        executorForCommand(command).execute(command);
    }

    private CommandExecutor executorForCommand(final BaseCommand command) {
        final CommandExecutor executor = commandExecutorMap.get(command.getCommand());
        if (executor != null) {
            return executor;
        }
        return NoopCommandExecutor.INSTANCE;
    }
}
  • CommandExecutorService实现了BootService, CommandExecutor接口,其execute方法通过executorForCommand从commandExecutorMap获取指定command对应的CommandExecutor,如果找不到则返回NoopCommandExecutor.INSTANCE

NoopCommandExecutor

skywalking-6.6.0/apm-sniffer/apm-agent-core/old-src/main/java/org/apache/skywalking/apm/agent/core/commands/executor/NoopCommandExecutor.java

public enum NoopCommandExecutor implements CommandExecutor {
    INSTANCE;

    @Override
    public void execute(final BaseCommand command) throws CommandExecutionException {

    }

}
  • NoopCommandExecutor实现了CommandExecutor接口,其execute方法为空操作

ServiceResetCommandExecutor

skywalking-6.6.0/apm-sniffer/apm-agent-core/old-src/main/java/org/apache/skywalking/apm/agent/core/commands/executor/ServiceResetCommandExecutor.java

public class ServiceResetCommandExecutor implements CommandExecutor {
    private static final ILog LOGGER = LogManager.getLogger(ServiceResetCommandExecutor.class);

    @Override
    public void execute(final BaseCommand command) throws CommandExecutionException {
        LOGGER.warn(&quot;Received ServiceResetCommand, a re-register task is scheduled.&quot;);

        ServiceManager.INSTANCE.findService(ServiceAndEndpointRegisterClient.class).coolDown();

        RemoteDownstreamConfig.Agent.SERVICE_ID = DictionaryUtil.nullValue();
        RemoteDownstreamConfig.Agent.SERVICE_INSTANCE_ID = DictionaryUtil.nullValue();
        RemoteDownstreamConfig.Agent.INSTANCE_REGISTERED_TIME = DictionaryUtil.nullValue();

        NetworkAddressDictionary.INSTANCE.clear();
        EndpointNameDictionary.INSTANCE.clear();
    }
}
  • ServiceResetCommandExecutor实现了CommandExecutor接口,其execute方法首先执行ServiceManager.INSTANCE.findService(ServiceAndEndpointRegisterClient.class).coolDown(),然后重置RemoteDownstreamConfig.Agent.SERVICE_ID、RemoteDownstreamConfig.Agent.SERVICE_INSTANCE_ID、RemoteDownstreamConfig.Agent.INSTANCE_REGISTERED_TIME,最后执行NetworkAddressDictionary.INSTANCE.clear()及EndpointNameDictionary.INSTANCE.clear()

小结

CommandService实现了BootService接口及Runnable接口,其boot方法会往executorService注册自己的run方法;其shutdown方法会执行commands.drainTo(new ArrayList<BaseCommand>())及executorService.shutdown();run方法会循环从commands取出command,然后通过serialNumberCache判断是否已经执行,已经执行则跳过,否则通过commandExecutorService.execute(command)执行并添加到serialNumberCache中;它还提供了receiveCommand方法,该方法会通过commands.offer(baseCommand)将command添加到commands队列

来源:35分类目录

声明:该文章系转载,转载该文章的目的在于更广泛的传递信息,并不代表本网站赞同其观点,文章内容仅供参考。

本站是一个个人学习和交流平台,网站上部分文章为网站管理员和网友从相关媒体转载而来,并不用于任何商业目的,内容为作者个人观点, 并不代表本网站赞同其观点和对其真实性负责。

我们已经尽可能的对作者和来源进行了通告,但是可能由于能力有限或疏忽,导致作者和来源有误,亦可能您并不期望您的作品在我们的网站上发布。我们为这些问题向您致歉,如果您在我站上发现此类问题,请及时联系我们,我们将根据您的要求,立即更正或者删除有关内容。本站拥有对此声明的最终解释权。