序
本文主要研究一下 nacos 的 TcpSuperSenseProcessor
TcpSuperSenseProcessor
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/TcpSuperSenseProcessor.java
@Component
public class TcpSuperSenseProcessor implements HealthCheckProcessor, Runnable {
@Autowired
private HealthCheckCommon healthCheckCommon;
@Autowired
private SwitchDomain switchDomain;
public static final int CONNECT_TIMEOUT_MS = 500;
private Map<String, BeatKey> keyMap = new ConcurrentHashMap<>();
private BlockingQueue<Beat> taskQueue = new LinkedBlockingQueue<Beat>();
/**
* this value has been carefully tuned, do not modify unless you're confident
*/
private static final int NIO_THREAD_COUNT = Runtime.getRuntime().availableProcessors() <= 1 ?
1 : Runtime.getRuntime().availableProcessors() / 2;
/**
* because some hosts doesn't support keep-alive connections, disabled temporarily
*/
private static final long TCP_KEEP_ALIVE_MILLIS = 0;
private static ScheduledExecutorService TCP_CHECK_EXECUTOR
= new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {Thread t = new Thread(r);
t.setName("nacos.naming.tcp.check.worker");
t.setDaemon(true);
return t;
}
});
private static ScheduledExecutorService NIO_EXECUTOR
= Executors.newScheduledThreadPool(NIO_THREAD_COUNT,
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {Thread thread = new Thread(r);
thread.setDaemon(true);
thread.setName("nacos.supersense.checker");
return thread;
}
}
);
private Selector selector;
public TcpSuperSenseProcessor() {
try {selector = Selector.open();
TCP_CHECK_EXECUTOR.submit(this);
} catch (Exception e) {throw new IllegalStateException("Error while initializing SuperSense(TM).");
}
}
@Override
public void process(HealthCheckTask task) {List<Instance> ips = task.getCluster().allIPs(false);
if (CollectionUtils.isEmpty(ips)) {return;}
for (Instance ip : ips) {if (ip.isMarked()) {if (SRV_LOG.isDebugEnabled()) {SRV_LOG.debug("tcp check, ip is marked as to skip health check, ip:" + ip.getIp());
}
continue;
}
if (!ip.markChecking()) {
SRV_LOG.warn("tcp check started before last one finished, service:"
+ task.getCluster().getService().getName() + ":"
+ task.getCluster().getName() + ":"
+ ip.getIp() + ":"
+ ip.getPort());
healthCheckCommon.reEvaluateCheckRT(task.getCheckRTNormalized() * 2, task, switchDomain.getTcpHealthParams());
continue;
}
Beat beat = new Beat(ip, task);
taskQueue.add(beat);
MetricsMonitor.getTcpHealthCheckMonitor().incrementAndGet();
}
}
private void processTask() throws Exception {Collection<Callable<Void>> tasks = new LinkedList<>();
do {Beat beat = taskQueue.poll(CONNECT_TIMEOUT_MS / 2, TimeUnit.MILLISECONDS);
if (beat == null) {return;}
tasks.add(new TaskProcessor(beat));
} while (taskQueue.size() > 0 && tasks.size() < NIO_THREAD_COUNT * 64);
for (Future<?> f : NIO_EXECUTOR.invokeAll(tasks)) {f.get();
}
}
@Override
public void run() {while (true) {
try {processTask();
int readyCount = selector.selectNow();
if (readyCount <= 0) {continue;}
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {SelectionKey key = iter.next();
iter.remove();
NIO_EXECUTOR.execute(new PostProcessor(key));
}
} catch (Throwable e) {SRV_LOG.error("[HEALTH-CHECK] error while processing NIO task", e);
}
}
}
//......
@Override
public String getType() {return "TCP";}
}- TcpSuperSenseProcessor 实现了 HealthCheckProcessor、Runnable 接口
- 其 process 方法会遍历 instances，对于非 markChecking 的会执行 healthCheckCommon.reEvaluateCheckRT，对于 marked 的直接跳过，对于 markChecking 的会创建 Beat 添加到 taskQueue
- 其构造器会往 TCP_CHECK_EXECUTOR 注册自己的 Runnable，其 run 方法不断执行 processTask 方法，然后从 selector 中 select key 然后创建 PostProcessor 提交给 NIO_EXECUTOR；processTask 方法会从 taskQueue 取出 Beat，然后创建 TaskProcessor 添加到 tasks，当 tasks 大小达到一定值则使用 NIO_EXECUTOR.invokeAll(tasks) 批量异步执行
PostProcessor
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/TcpSuperSenseProcessor.java
public class PostProcessor implements Runnable {
SelectionKey key;
public PostProcessor(SelectionKey key) {this.key = key;}
@Override
public void run() {Beat beat = (Beat) key.attachment();
SocketChannel channel = (SocketChannel) key.channel();
try {if (!beat.isHealthy()) {
//invalid beat means this server is no longer responsible for the current service
key.cancel();
key.channel().close();
beat.finishCheck();
return;
}
if (key.isValid() && key.isConnectable()) {
//connected
channel.finishConnect();
beat.finishCheck(true, false, System.currentTimeMillis() - beat.getTask().getStartTime(), "tcp:ok+");
}
if (key.isValid() && key.isReadable()) {
//disconnected
ByteBuffer buffer = ByteBuffer.allocate(128);
if (channel.read(buffer) == -1) {key.cancel();
key.channel().close();
} else {// not terminate request, ignore}
}
} catch (ConnectException e) {
// unable to connect, possibly port not opened
beat.finishCheck(false, true, switchDomain.getTcpHealthParams().getMax(), "tcp:unable2connect:" + e.getMessage());
} catch (Exception e) {beat.finishCheck(false, false, switchDomain.getTcpHealthParams().getMax(), "tcp:error:" + e.getMessage());
try {key.cancel();
key.channel().close();
} catch (Exception ignore) {}}
}
}- PostProcessor 实现了 Runnable 接口，其 run 方法主要是执行 beat.finishCheck
TaskProcessor
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/TcpSuperSenseProcessor.java
private class TaskProcessor implements Callable<Void> {
private static final int MAX_WAIT_TIME_MILLISECONDS = 500;
Beat beat;
public TaskProcessor(Beat beat) {this.beat = beat;}
@Override
public Void call() {long waited = System.currentTimeMillis() - beat.getStartTime();
if (waited > MAX_WAIT_TIME_MILLISECONDS) {Loggers.SRV_LOG.warn("beat task waited too long:" + waited + "ms");
}
SocketChannel channel = null;
try {Instance instance = beat.getIp();
Cluster cluster = beat.getTask().getCluster();
BeatKey beatKey = keyMap.get(beat.toString());
if (beatKey != null && beatKey.key.isValid()) {if (System.currentTimeMillis() - beatKey.birthTime < TCP_KEEP_ALIVE_MILLIS) {instance.setBeingChecked(false);
return null;
}
beatKey.key.cancel();
beatKey.key.channel().close();
}
channel = SocketChannel.open();
channel.configureBlocking(false);
// only by setting this can we make the socket close event asynchronous
channel.socket().setSoLinger(false, -1);
channel.socket().setReuseAddress(true);
channel.socket().setKeepAlive(true);
channel.socket().setTcpNoDelay(true);
int port = cluster.isUseIPPort4Check() ? instance.getPort() : cluster.getDefCkport();
channel.connect(new InetSocketAddress(instance.getIp(), port));
SelectionKey key
= channel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
key.attach(beat);
keyMap.put(beat.toString(), new BeatKey(key));
beat.setStartTime(System.currentTimeMillis());
NIO_EXECUTOR.schedule(new TimeOutTask(key),
CONNECT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
} catch (Exception e) {beat.finishCheck(false, false, switchDomain.getTcpHealthParams().getMax(), "tcp:error:" + e.getMessage());
if (channel != null) {
try {channel.close();
} catch (Exception ignore) {}}
}
return null;
}
}- TaskProcessor 实现了 Callable<Void> 接口，其 call 方法主要是对目标 instance 执行 beat 操作，同时它会往 NIO_EXECUTOR 注册 TimeOutTask 的延时任务
TimeOutTask
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/TcpSuperSenseProcessor.java
private static class TimeOutTask implements Runnable {
SelectionKey key;
public TimeOutTask(SelectionKey key) {this.key = key;}
@Override
public void run() {if (key != null && key.isValid()) {SocketChannel channel = (SocketChannel) key.channel();
Beat beat = (Beat) key.attachment();
if (channel.isConnected()) {return;}
try {channel.finishConnect();
} catch (Exception ignore) { }
try {beat.finishCheck(false, false, beat.getTask().getCheckRTNormalized() * 2, "tcp:timeout");
key.cancel();
key.channel().close();
} catch (Exception ignore) {}}
}
}- TimeOutTask 实现了 Runnable 方法，其 run 方法会执行 channel.finishConnect()，然后执行 beat.finishCheck 标记 success 为 false，msg 为 tcp:timeout
小结
- TcpSuperSenseProcessor 实现了 HealthCheckProcessor、Runnable 接口
- 其 process 方法会遍历 instances，对于非 markChecking 的会执行 healthCheckCommon.reEvaluateCheckRT，对于 marked 的直接跳过，对于 markChecking 的会创建 Beat 添加到 taskQueue
- 其构造器会往 TCP_CHECK_EXECUTOR 注册自己的 Runnable，其 run 方法不断执行 processTask 方法，然后从 selector 中 select key 然后创建 PostProcessor 提交给 NIO_EXECUTOR；processTask 方法会从 taskQueue 取出 Beat，然后创建 TaskProcessor 添加到 tasks，当 tasks 大小达到一定值则使用 NIO_EXECUTOR.invokeAll(tasks) 批量异步执行
doc
- TcpSuperSenseProcessor