sentinel-transport-SPI-HeartbeatSenderInitFunc

说明

我们引入以下依赖

<dependency><groupId>com.alibaba.csp</groupId><artifactId>sentinel-transport-simple-http</artifactId><version>1.8.6</version>
</dependency>

配置环境变量

-Dcsp.sentinel.dashboard.server=localhost:8080

则会上报到dashboard,如果关闭服务，一段时间时候就会自动剔除,判断是否失联则是根据HeartbeatSenderInitFunc在定期发送心跳的

 

通过初始化com.alibaba.csp.sentinel.init.InitExecutor#doInit 之后，

<1>

com.alibaba.csp.sentinel.init.InitExecutor#doInit

 public static void doInit() {//原子性set和判断是否已经初始化if (!initialized.compareAndSet(false, true)) {return;}try {//SPI获取InitFuncServiceLoader<InitFunc> loader = ServiceLoaderUtil.getServiceLoader(InitFunc.class);List<OrderWrapper> initList = new ArrayList<OrderWrapper>();for (InitFunc initFunc : loader) {RecordLog.info("[InitExecutor] Found init func: " + initFunc.getClass().getCanonicalName());insertSorted(initList, initFunc);}for (OrderWrapper w : initList) {//<1-2>com.alibaba.csp.sentinel.transport.init.HeartbeatSenderInitFunc#init
                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();}}

<1-2>

com.alibaba.csp.sentinel.transport.init.HeartbeatSenderInitFunc#init

    @Overridepublic void init() {//这里也是通过SPI获取HeartbeatSender 我们可以自定义HeartbeatSender sender = HeartbeatSenderProvider.getHeartbeatSender();if (sender == null) {RecordLog.warn("[HeartbeatSenderInitFunc] WARN: No HeartbeatSender loaded");return;}//初始化ScheduledThreadPoolExecutor 通过他来完成定时心跳发送
        initSchedulerIfNeeded();//获取心跳间隔时间 我们可以通过环境变量csp.sentinel.heartbeat.interval.ms long interval = retrieveInterval(sender);//设置到config 后去使用
        setIntervalIfNotExists(interval);//<1-2-1>开启心跳定时发送任务
        scheduleHeartbeatTask(sender, interval);}

<1-2-1->

com.alibaba.csp.sentinel.transport.init.HeartbeatSenderInitFunc#scheduleHeartbeatTask

 private void scheduleHeartbeatTask(/*@NonNull*/ final HeartbeatSender sender, /*@Valid*/ long interval) {pool.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {//<1-2-1-1>发送心跳
                    sender.sendHeartbeat();} catch (Throwable e) {RecordLog.warn("[HeartbeatSender] Send heartbeat error", e);}}}, 5000, interval, TimeUnit.MILLISECONDS);RecordLog.info("[HeartbeatSenderInit] HeartbeatSender started: "+ sender.getClass().getCanonicalName());}

<1-2-1-1>

 public boolean sendHeartbeat() throws Exception {if (TransportConfig.getRuntimePort() <= 0) {RecordLog.info("[SimpleHttpHeartbeatSender] Command server port not initialized, won't send heartbeat", new Object[0]);return false;} else {//获取dashboard地址 key=ip value=端口Tuple2<String, Integer> addrInfo = this.getAvailableAddress();if (addrInfo == null) {return false;} else {InetSocketAddress addr = new InetSocketAddress((String)addrInfo.r1, (Integer)addrInfo.r2);//获取心跳地址可以通过csp.sentinel.heartbeat.api.path指定 默认是 /registry/machineSimpleHttpRequest request = new SimpleHttpRequest(addr, TransportConfig.getHeartbeatApiPath());request.setParams(this.heartBeat.generateCurrentMessage());try {//http请求发送心跳SimpleHttpResponse response = this.httpClient.post(request);if (response.getStatusCode() == 200) {return true;}if (this.clientErrorCode(response.getStatusCode()) || this.serverErrorCode(response.getStatusCode())) {RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr + ", http status code: " + response.getStatusCode(), new Object[0]);}} catch (Exception var5) {RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr, var5);}return false;}}}