聊聊skywaking的CommandService
2020/3/1 17:15:09
本文主要是介绍聊聊skywaking的CommandService,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!
序
本文主要研究一下skywalking的CommandService
CommandService
skywalking-6.6.0/apm-sniffer/apm-agent-core/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())及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/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<String> queue; private final int maxCapacity; public CommandSerialNumberCache() { this(DEFAULT_MAX_CAPACITY); } public CommandSerialNumberCache(int maxCapacity) { queue = new LinkedBlockingDeque<String>(maxCapacity); this.maxCapacity = maxCapacity; } public void add(String number) { if (queue.size() >= 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/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/src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandExecutorService.java
@DefaultImplementor public class CommandExecutorService implements BootService, CommandExecutor { private Map<String, CommandExecutor> commandExecutorMap; @Override public void prepare() throws Throwable { commandExecutorMap = new HashMap<String, CommandExecutor>(); // 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/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/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("Received ServiceResetCommand, a re-register task is scheduled."); 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())及executorService.shutdown();run方法会循环从commands取出command,然后通过serialNumberCache判断是否已经执行,已经执行则跳过,否则通过commandExecutorService.execute(command)执行并添加到serialNumberCache中;它还提供了receiveCommand方法,该方法会通过commands.offer(baseCommand)将command添加到commands队列
doc
这篇关于聊聊skywaking的CommandService的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!
- 2024-10-04el-table 开启定时器下,表格的选中状态会消失是什么原因-icode9专业技术文章分享
- 2024-10-03如何安装和初始化飞牛私有云 fnOS?-icode9专业技术文章分享
- 2024-10-03如何安装 App 并连接到飞牛 NAS?-icode9专业技术文章分享
- 2024-10-03如何安装飞牛 TV 并连接到影视服务器?-icode9专业技术文章分享
- 2024-10-03如何在PVE和ESXI上安装飞牛私有云 fnOS?-icode9专业技术文章分享
- 2024-10-03fnOS国产最强NAS安装系统异常情况处理-icode9专业技术文章分享
- 2024-10-03飞牛NAS如何创建存储空间?-icode9专业技术文章分享
- 2024-10-03fnOS国产最强NAS硬盘会自动休眠吗?-icode9专业技术文章分享
- 2024-10-03fnOS国产最强NAS如何安装飞牛影视和创建媒体库?-icode9专业技术文章分享
- 2024-10-03fnOS国产最强NAS如何为家人朋友开通影视账号?-icode9专业技术文章分享