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. 安裝cuda+cuDNN
- 2. GitHub的使用說明
- 3. phpDocumentor使用教程【安裝PHPDocumentor】
- 4. yarn run build報錯Component is not found in path 「npm/taro-ui/dist/weapp/components/rate/index「
- 5. 精講Haproxy搭建Web集羣
- 6. 安全測試基礎之MySQL
- 7. C/C++編程筆記：C語言中的複雜聲明分析，用實例帶你完全讀懂
- 8. Python3教程(1)----搭建Python環境
- 9. 李宏毅機器學習課程筆記2：Classification、Logistic Regression、Brief Introduction of Deep Learning
- 10. 阿里雲ECS配置速記
- 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的初始化