序
本文主要研究一下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<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/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<BaseCommand>())及executorService.shutdown()；run方法会循环从commands取出command，然后通过serialNumberCache判断是否已经执行，已经执行则跳过，否则通过commandExecutorService.execute(command)执行并添加到serialNumberCache中；它还提供了receiveCommand方法，该方法会通过commands.offer(baseCommand)将command添加到commands队列
