kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Make `CommandService` daemon to avoid blocking target application shutting down gracefully (#4541)

This commit is contained in:
kezhenxu94 2020-03-19 13:53:54 +08:00 committed by GitHub
parent 28ac73bff3
commit 967ed8f774
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 17 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/commands/CommandService.java
View File

@ -17,8 +17,13 @@
package org.apache.skywalking.apm.agent.core.commands;
import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.skywalking.apm.agent.core.boot.BootService;
import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor;
import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory;
import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
import org.apache.skywalking.apm.agent.core.logging.api.ILog;
import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
@ -29,19 +34,16 @@ import org.apache.skywalking.apm.network.trace.component.command.CommandDeserial
import org.apache.skywalking.apm.network.trace.component.command.UnsupportedCommandException;
import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
@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 ExecutorService executorService = Executors.newSingleThreadExecutor(
new DefaultNamedThreadFactory("CommandService")
);
private LinkedBlockingQueue<BaseCommand> commands = new LinkedBlockingQueue<>(64);
private CommandSerialNumberCache serialNumberCache = new CommandSerialNumberCache();
@Override
@ -50,12 +52,9 @@ public class CommandService implements BootService, Runnable {
@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");
}
}));
executorService.submit(
new RunnableWithExceptionProtection(this, t -> LOGGER.error(t, "CommandService failed to execute commands"))
);
}
@Override
@ -94,7 +93,7 @@ public class CommandService implements BootService, Runnable {
@Override
public void shutdown() throws Throwable {
isRunning = false;
commands.drainTo(new ArrayList<BaseCommand>());
commands.drainTo(new ArrayList<>());
executorService.shutdown();
}
@ -111,8 +110,10 @@ public class CommandService implements BootService, Runnable {
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());
LOGGER.warn(
"Command[{}, {}] cannot add to command list. because the command list is full.",
baseCommand.getCommand(), baseCommand.getSerialNumber()
);
}
} catch (UnsupportedCommandException e) {
if (LOGGER.isWarnEnable()) {