7.Sentinel源码分析—Sentinel是怎么和控制台通讯的？
这里会介绍：html
- Sentinel会使用多线程的方式实现一个类Reactor的IO模型
- Sentinel会使用心跳检测来观察控制台是否正常
Sentinel源码解析系列：java
1.Sentinel源码分析—FlowRuleManager加载规则作了什么？node
2. Sentinel源码分析—Sentinel是如何进行流量统计的？react
3. Sentinel源码分析— QPS流量控制是如何实现的？git
4.Sentinel源码分析— Sentinel是如何作到降级的？github
5.Sentinel源码分析—Sentinel如何实现自适应限流？c#
6.Sentinel源码分析—Sentinel是如何动态加载配置限流的？多线程
在看个人这篇文章以前你们能够先看一下官方的这篇文章：github.com/alibaba/Sen…app
客户端会在InitExecutor调用doInit方法中与控制台创建通讯，因此咱们直接看doInit方法：异步
InitExecutor#doInit
public static void doInit() {
//InitExecutor只会初始化一次，而且初始化失败会退出
if (!initialized.compareAndSet(false, true)) {
return;
}
try {
//经过spi加载InitFunc子类
ServiceLoader<InitFunc> loader = ServiceLoader.load(InitFunc.class);
List<OrderWrapper> initList = new ArrayList<OrderWrapper>();
for (InitFunc initFunc : loader) {
RecordLog.info("[InitExecutor] Found init func: " + initFunc.getClass().getCanonicalName());
//给全部的initFunc排序，按@InitOrder从小到大进行排序
//而后封装成OrderWrapper对象
insertSorted(initList, initFunc);
}
for (OrderWrapper w : initList) {
w.func.init();
RecordLog.info(String.format("[InitExecutor] Executing %s with order %d",
w.func.getClass().getCanonicalName(), w.order));
}
} catch (Exception ex) {
RecordLog.warn("[InitExecutor] WARN: Initialization failed", ex);
ex.printStackTrace();
} catch (Error error) {
RecordLog.warn("[InitExecutor] ERROR: Initialization failed with fatal error", error);
error.printStackTrace();
}
}
复制代码
由于这里咱们引入了sentinel-transport-simple-http模块，因此使用spi加载InitFunc的子类的时候会加载三个子类实例，分别是：CommandCenterInitFunc、HeartbeatSenderInitFunc、MetricCallbackInit。 而后会遍历loader，根据@InitOrder的大小进行排序，并封装成OrderWrapper放入到initList中。 因此initList里面的对象顺序是：
- CommandCenterInitFunc
- HeartbeatSenderInitFunc
- MetricCallbackInit 而后遍历initList依次调用init方法。
因此下面咱们来看一下这三个实现类的init方法作了什么：
CommandCenterInitFunc
CommandCenterInitFunc#init
public void init() throws Exception {
//获取commandCenter对象
CommandCenter commandCenter = CommandCenterProvider.getCommandCenter();
if (commandCenter == null) {
RecordLog.warn("[CommandCenterInitFunc] Cannot resolve CommandCenter");
return;
}
//调用SimpleHttpCommandCenter的beforeStart方法
//用来设置CommandHandler的实现类
commandCenter.beforeStart();
commandCenter.start();
RecordLog.info("[CommandCenterInit] Starting command center: "
+ commandCenter.getClass().getCanonicalName());
}
复制代码
这个方法里面的全部操做都是针对CommandCenter来进行的，因此咱们先来看看CommandCenterProvider这个类。
CommandCenterProvider
static {
//初始化commandCenter对象
resolveInstance();
}
private static void resolveInstance() {
//获取SpiOrder更大的子类实现类
CommandCenter resolveCommandCenter = SpiLoader.loadHighestPriorityInstance(CommandCenter.class);
if (resolveCommandCenter == null) {
RecordLog.warn("[CommandCenterProvider] WARN: No existing CommandCenter found");
} else {
commandCenter = resolveCommandCenter;
RecordLog.info("[CommandCenterProvider] CommandCenter resolved: " + resolveCommandCenter.getClass()
.getCanonicalName());
}
}
复制代码
CommandCenterProvider会在首次初始化的时候调用resolveInstance方法。在resolveInstance方法里面会调用SpiLoader.loadHighestPriorityInstance来获取CommandCenter，这里获取的是SimpleHttpCommandCenter这个实例，loadHighestPriorityInstance方法具体的实现很是简单，我就不去分析了。 而后将commandCenter赋值SimpleHttpCommandCenter实例。
因此CommandCenterProvider.getCommandCenter()方法返回的是SimpleHttpCommandCenter实例。
而后调用SimpleHttpCommandCenter的beforeStart方法。
SimpleHttpCommandCenter#beforeStart
public void beforeStart() throws Exception {
// Register handlers
//调用CommandHandlerProvider的namedHandlers方法
//获取CommandHandler的spi中设置的实现类
Map<String, CommandHandler> handlers = CommandHandlerProvider.getInstance().namedHandlers();
//将handlers中的数据设置到handlerMap中
registerCommands(handlers);
}
复制代码
这个方法首先会调用CommandHandlerProvider的namedHandlers中获取全部的CommandHandler实现类。
CommandHandlerProvider#namedHandlers
private final ServiceLoader<CommandHandler> serviceLoader = ServiceLoader.load(CommandHandler.class);
public Map<String, CommandHandler> namedHandlers() {
Map<String, CommandHandler> map = new HashMap<String, CommandHandler>();
for (CommandHandler handler : serviceLoader) {
//获取实现类CommandMapping注解的name属性
String name = parseCommandName(handler);
if (!StringUtil.isEmpty(name)) {
map.put(name, handler);
}
}
return map;
}
复制代码
这个类会经过spi先加载CommandHandler的实现类，而后将实现类按注解上面的name属性放入到map里面去。 CommandHandler的实现类是用来和控制台进行交互的处理类，负责处理。 这也是策略模式的一种应用，根据map里面的不一样策略来作不一样的处理，例如SendMetricCommandHandler是用来统计调用信息而后发送给控制台用的，ModifyRulesCommandHandler是用来作实时修改限流策略的处理的等等。
而后咱们再回到CommandCenterInitFunc中，继续往下走，调用commandCenter.start()方法。
SimpleHttpCommandCenter#start
public void start() throws Exception {
//获取当前机器的cpu线程数
int nThreads = Runtime.getRuntime().availableProcessors();
//建立一个cpu线程数大小的固定线程池，用来作业务线程池用
this.bizExecutor = new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<Runnable>(10),
new NamedThreadFactory("sentinel-command-center-service-executor"),
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
CommandCenterLog.info("EventTask rejected");
throw new RejectedExecutionException();
}
});
Runnable serverInitTask = new Runnable() {
int port;
{
try {
//获取port
port = Integer.parseInt(TransportConfig.getPort());
} catch (Exception e) {
port = DEFAULT_PORT;
}
}
@Override
public void run() {
boolean success = false;
//建立一个ServerSocket
ServerSocket serverSocket = getServerSocketFromBasePort(port);
if (serverSocket != null) {
CommandCenterLog.info("[CommandCenter] Begin listening at port " + serverSocket.getLocalPort());
socketReference = serverSocket;
executor.submit(new ServerThread(serverSocket));
success = true;
port = serverSocket.getLocalPort();
} else {
CommandCenterLog.info("[CommandCenter] chooses port fail, http command center will not work");
}
if (!success) {
port = PORT_UNINITIALIZED;
}
TransportConfig.setRuntimePort(port);
//关闭线程池
executor.shutdown();
}
};
new Thread(serverInitTask).start();
}
复制代码
- 这个方法会建立一个固定大小的业务线程池
- 建立一个serverInitTask，里面负责创建serverSocket而后用executor去建立一个ServerThread异步执行serverSocket
- executor用完以后会在serverInitTask里面调用executor的shutdown方法去关闭线程池
其中executor是一个单线程的线程池：
private ExecutorService executor = Executors.newSingleThreadExecutor(
new NamedThreadFactory("sentinel-command-center-executor"));
复制代码
ServerThread是SimpleHttpCommandCenter的内部类：
public void run() {
while (true) {
Socket socket = null;
try {
//创建链接
socket = this.serverSocket.accept();
//默认的超时时间是3s
setSocketSoTimeout(socket);
HttpEventTask eventTask = new HttpEventTask(socket);
//使用业务线程异步处理
bizExecutor.submit(eventTask);
} catch (Exception e) {
CommandCenterLog.info("Server error", e);
if (socket != null) {
try {
socket.close();
} catch (Exception e1) {
CommandCenterLog.info("Error when closing an opened socket", e1);
}
}
try {
// In case of infinite log.
Thread.sleep(10);
} catch (InterruptedException e1) {
// Indicates the task should stop.
break;
}
}
}
}
复制代码
run方法会使用构造器传入的serverSocket创建链接后设置超时时间，封装成HttpEventTask类，而后使用上面建立的bizExecutor异步执行任务。
HttpEventTask是Runnable的实现类，因此调用bizExecutor的submit的时候会调用其中的run方法使用socket与控制台进行交互。
HttpEventTask#run
public void run() {
....
// Validate the target command.
//获取commandName
String commandName = HttpCommandUtils.getTarget(request);
if (StringUtil.isBlank(commandName)) {
badRequest(printWriter, "Invalid command");
return;
}
// Find the matching command handler.
//根据commandName获取处理器名字
CommandHandler<?> commandHandler = SimpleHttpCommandCenter.getHandler(commandName);
if (commandHandler != null) {
//调用处理器结果，而后返回给控制台
CommandResponse<?> response = commandHandler.handle(request);
handleResponse(response, printWriter, outputStream);
}
....
} catch (Throwable e) {
....
} finally {
....
}
}
复制代码
HttpEventTask的run方法很长，可是不少都是有关输入输出流的，咱们不关心，因此省略。只须要知道会把request请求最后转换成一个控制台发过来的指令，而后经过SimpleHttpCommandCenter调用getHandler获得处理器，而后处理数据就好了。
因此这个整个的处理流程就是：
SimpleHttpCommandCenter#getHandler
public static CommandHandler getHandler(String commandName) {
return handlerMap.get(commandName);
}
复制代码
handlerMap里面的数据是经过前面咱们分析的调用beforeStart方法设置进来的。
而后经过commandName获取对应的控制台，例如：控制台发送过来metric指令，那么就会对应的调用SendMetricCommandHandler的handle方法来处理控制台的指令。
咱们来看看SendMetricCommandHandler是怎么处理返回统计数据的：
SendMetricCommandHandler#handle
public CommandResponse<String> handle(CommandRequest request) {
// Note: not thread-safe.
if (searcher == null) {
synchronized (lock) {
//获取应用名
String appName = SentinelConfig.getAppName();
if (appName == null) {
appName = "";
}
if (searcher == null) {
//用来找metric文件，
searcher = new MetricSearcher(MetricWriter.METRIC_BASE_DIR,
MetricWriter.formMetricFileName(appName, PidUtil.getPid()));
}
}
}
//获取请求的开始结束时间和最大的行数
String startTimeStr = request.getParam("startTime");
String endTimeStr = request.getParam("endTime");
String maxLinesStr = request.getParam("maxLines");
//用来肯定资源
String identity = request.getParam("identity");
long startTime = -1;
int maxLines = 6000;
if (StringUtil.isNotBlank(startTimeStr)) {
startTime = Long.parseLong(startTimeStr);
} else {
return CommandResponse.ofSuccess("");
}
List<MetricNode> list;
try {
// Find by end time if set.
if (StringUtil.isNotBlank(endTimeStr)) {
long endTime = Long.parseLong(endTimeStr);
//根据开始结束时间找到统计数据
list = searcher.findByTimeAndResource(startTime, endTime, identity);
} else {
if (StringUtil.isNotBlank(maxLinesStr)) {
maxLines = Integer.parseInt(maxLinesStr);
}
maxLines = Math.min(maxLines, 12000);
list = searcher.find(startTime, maxLines);
}
} catch (Exception ex) {
return CommandResponse.ofFailure(new RuntimeException("Error when retrieving metrics", ex));
}
if (list == null) {
list = new ArrayList<>();
}
//若是identity为空就加入CPU负载和系统负载
if (StringUtil.isBlank(identity)) {
addCpuUsageAndLoad(list);
}
StringBuilder sb = new StringBuilder();
for (MetricNode node : list) {
sb.append(node.toThinString()).append("\n");
}
return CommandResponse.ofSuccess(sb.toString());
}
复制代码
咱们在1.Sentinel源码分析—FlowRuleManager加载规则作了什么？里介绍了Metric统计信息会在MetricTimerListener的run方法中定时写入文件中去。
因此handle方法里面主要是如何根据请求的开始结束时间，资源名来获取磁盘的文件，而后返回磁盘的统计信息，并记录一下当前的统计信息，防止重复发送统计数据到控制台。
HeartbeatSenderInitFunc
HeartbeatSenderInitFunc主要是用来作心跳线程使用的，按期的和控制台进行心跳链接。
HeartbeatSenderInitFunc#init
public void init() {
//获取HeartbeatSender的实现类
HeartbeatSender sender = HeartbeatSenderProvider.getHeartbeatSender();
if (sender == null) {
RecordLog.warn("[HeartbeatSenderInitFunc] WARN: No HeartbeatSender loaded");
return;
}
//建立一个corepoolsize为2，maximumPoolSize为最大的线程池
initSchedulerIfNeeded();
//获取心跳间隔时间，默认10s
long interval = retrieveInterval(sender);
//设置间隔心跳时间
setIntervalIfNotExists(interval);
//开启一个定时任务，每隔interval时间发送一个心跳
scheduleHeartbeatTask(sender, interval);
}
复制代码
- 首先会调用HeartbeatSenderProvider.getHeartbeatSender方法，里面会根据spi建立实例，返回一个SimpleHttpHeartbeatSender实例。
- 调用initSchedulerIfNeeded方法建立一个corepoolsize为2的线程池
- 获取心跳间隔时间，若是没有设置，那么是10s
- 调用scheduleHeartbeatTask方法开启一个定时线程调用。
咱们来看看scheduleHeartbeatTask方法： HeartbeatSenderInitFunc#scheduleHeartbeatTask
private void scheduleHeartbeatTask(/*@NonNull*/ final HeartbeatSender sender, /*@Valid*/ long interval) {
pool.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
sender.sendHeartbeat();
} catch (Throwable e) {
RecordLog.warn("[HeartbeatSender] Send heartbeat error", e);
}
}
}, 5000, interval, TimeUnit.MILLISECONDS);
RecordLog.info("[HeartbeatSenderInit] HeartbeatSender started: "
+ sender.getClass().getCanonicalName());
}
复制代码
默认的状况，建立的这个定时任务会每隔10s调用一次SimpleHttpHeartbeatSender的sendHeartbeat方法。
SimpleHttpHeartbeatSender#sendHeartbeat
public boolean sendHeartbeat() throws Exception {
if (TransportConfig.getRuntimePort() <= 0) {
RecordLog.info("[SimpleHttpHeartbeatSender] Runtime port not initialized, won't send heartbeat");
return false;
}
//获取控制台的ip和端口等信息
InetSocketAddress addr = getAvailableAddress();
if (addr == null) {
return false;
}
//设置http调用的ip和端口，还有访问的url
SimpleHttpRequest request = new SimpleHttpRequest(addr, HEARTBEAT_PATH);
//获取版本号，端口等信息
request.setParams(heartBeat.generateCurrentMessage());
try {
//发送post请求
SimpleHttpResponse response = httpClient.post(request);
if (response.getStatusCode() == OK_STATUS) {
return true;
}
} catch (Exception e) {
RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr + " : ", e);
}
return false;
}
复制代码
这个心跳检测的方法就写的很简单了，经过Dcsp.sentinel.dashboard.server预先设置好的ip和端口号发送post请求到控制台，而后检测是否返回200，若是是则说明控制台正常，不然进行异常处理。
- 1. Redis Sentinel 源码分析 - Sentinel 选举机制和主备倒换
- 2. 4.Sentinel源码分析— Sentinel是如何作到降级的？
- 3. 源码分析Sentinel专栏
- 4. Sentinel 源码解析
- 5. Redis Sentinel 源码分析 - Sentinel 网络连接和Tilt模式
- 6. RateLimiter 源码分析(Guava 和 Sentinel 实现)
- 7. Sentinel 实战-控制台篇
- 8. Sentinel学习(三) —— 控制台
- 9. Alibaba Sentinel 源码分析(6) NodeSelectorSlot 分析
- 10. Redis Sentinel 源码分析 - Sentinel的初始化
- 更多相关文章...
- • Hibernate是什么 - Hibernate教程
- • MyBatis是什么 - MyBatis教程
- • 互联网组织的未来：剖析GitHub员工的任性之源
- • 漫谈MySQL的锁机制
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. Redis Sentinel 源码分析 - Sentinel 选举机制和主备倒换
- 2. 4.Sentinel源码分析— Sentinel是如何作到降级的？
- 3. 源码分析Sentinel专栏
- 4. Sentinel 源码解析
- 5. Redis Sentinel 源码分析 - Sentinel 网络连接和Tilt模式
- 6. RateLimiter 源码分析(Guava 和 Sentinel 实现)
- 7. Sentinel 实战-控制台篇
- 8. Sentinel学习(三) —— 控制台
- 9. Alibaba Sentinel 源码分析(6) NodeSelectorSlot 分析
- 10. Redis Sentinel 源码分析 - Sentinel的初始化