文档章节

Azkaban的线程系列 37:QueueProcessor线程的任务处理&executor存活监控

强子1985
 强子1985
发布于 2016/04/13 13:32
字数 470
阅读 381
收藏 0

有个线程,AzkabanWebServer-QueueProcessor-Thread.

下面分析下这个线程到底干嘛的!!!

====================================================================================

stop in azkaban.executor.ExecutorManager$QueueProcessorThread.run

====================================================================================

public void run() {

// Loops till QueueProcessorThread is shutdown

while (!shutdown) {// 一直循环

synchronized (this) {//

try {

// start processing queue if active, other wait for

// sometime

if (isActive) {

// 开始处理任务

processQueuedFlows(activeExecutorRefreshWindowInMilisec,

activeExecutorRefreshWindowInFlows);

}

wait(QUEUE_PROCESSOR_WAIT_IN_MS);// 等待1秒

} catch (Exception e) {

logger.error("QueueProcessorThread Interrupted. Probably to shut down.", e);

}

}

}

}

所以接下来要看这个processQueuedFlows干嘛的?

====================================================================================

// process flow with current snapshot of activeExecutors

selectExecutorAndDispatchFlow(reference, exflow, new HashSet<Executor>(activeExecutors));

从这里来看,就是把任务分发到具体的executor上,

不过这个函数其实还做了其它的事情!那就是executor的存活监控

// if we have dispatched more than maxContinuousFlowProcessed or

// It has been more then activeExecutorsRefreshWindow millisec

// since we

// refreshed

// 满足上面的条件就探测executor的存活性

if (currentTime - lastExecutorRefreshTime > activeExecutorsRefreshWindow

|| currentContinuousFlowProcessed >= maxContinuousFlowProcessed) {

// Refresh executorInfo for all activeExecutors

refreshExecutors();

lastExecutorRefreshTime = currentTime;

currentContinuousFlowProcessed = 0;

}


所以接下来,我们来看看refreshExecutors的代码实现!

===================================================================================

jdb azkaban.webapp.AzkabanWebServer -conf  /root/azkb/azkaban_3.0.0_debug/conf

stop in  azkaban.executor.ExecutorManager.refreshExecutors

==============================================================================

其中刷新周期

azkProps.getLong(AZKABAN_ACTIVE_EXECUTOR_REFRESH_IN_MS, 50000),

private static final String AZKABAN_ACTIVE_EXECUTOR_REFRESH_IN_MS = "azkaban.activeexecutor.refresh.milisecinterval";

默认为50秒刷1次更新请求


其实是通过一个线程池来跑的,

Future<String> fetchExecutionInfo = executorInforRefresherService.submit(new Callable<String>() {

@Override

public String call() throws Exception {

return callExecutorForJsonString(executor.getHost(), executor.getPort(), "/serverStatistics",

null);

}

});

那么,这个线程池的个数呢?


executorInforRefresherService = Executors

.newFixedThreadPool(azkProps.getInt(AZKABAN_EXECUTORINFO_REFRESH_MAX_THREADS, 5));

private static final String AZKABAN_EXECUTORINFO_REFRESH_MAX_THREADS = "azkaban.executorinfo.refresh.maxThreads";

默认就是5个了,好,回来看怎么执行!

==============================================================================

发出HTTP请求 /serverStatistics

然后结果更新到executor,所以重点就是 /serverStatistics在executor中的执行过程!

==============================================================================

root.addServlet(new ServletHolder(new ServerStatisticsServlet()), "/serverStatistics");



jdb azkaban.execapp.AzkabanExecutorServer  -conf  /root/azkb/azkaban_3.0.0_debug/conf

stop in azkaban.execapp.ServerStatisticsServlet.doGet

stop in azkaban.execapp.ServerStatisticsServlet.populateStatistics


run

其实就是生成本地的信息汇总

1)

fillRemainingMemoryPercent(stats);

azkaban.execapp.ServerStatisticsServlet.fillRemainingMemoryPercent

 

2)

fillRemainingFlowCapacityAndLastDispatchedTime(stats);

stop in azkaban.execapp.ServerStatisticsServlet.fillRemainingFlowCapacityAndLastDispatchedTime

3)

fillCpuUsage(stats);

==============================================================================

其实就是返回这3种信息

对于web server来说,5秒内拿到返回内容后,更新本地消息

  public void setExecutorInfo(ExecutorInfo info) {

    this.cachedExecutorStats = info;

    this.lastStatsUpdatedTime = new Date();

  }

以此来作为存活监控!

© 著作权归作者所有

共有 人打赏支持
强子1985

强子1985

粉丝 862
博文 935
码字总数 645914
作品 8
南京
架构师
Azkaban的Web Server源码探究系列24: 一次性执行execute任务取出处理

提交之后,很不幸,有个问题如下: 抓包分析: 请求和响应分别如下: 所以接下来需要对这个URL进行debug ============================================================================== ...

强子哥哥
2016/04/10
212
0
工作流调度器Azkaban的安装配置

Azkaban安装部署 准备工作 Azkaban Web服务器 azkaban-web-server-2.5.0.tar.gz Azkaban执行服务器 azkaban-executor-server-2.5.0.tar.gz MySQL 目前azkaban只支持 mysql,需安装mysql服务器......

ChinaUnicom110
2017/09/29
0
0
java多线程系列:Executors框架

目录 Executor接口介绍 ExecutorService常用接口介绍 创建线程池的一些方法介绍 3.1 newFixedThreadPool方法 3.2 newCachedThreadPool方法 3.3 newScheduledThreadPool方法 疑问解答 4.1. R...

勿妄
06/12
0
0
java并发库之Executors常用的创建ExecutorService的几个方法说明

一、线程池的创建 我们可以通过ThreadPoolExecutor来创建一个线程池。 new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, milliseconds,runnableTaskQueue, handler);......

文星
2011/10/16
0
5
Executors常用的创建ExecutorService的方法的简单说明

一、线程池的创建 我们可以通过ThreadPoolExecutor来创建一个线程池。 Java代码 ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQ......

引鸩怼孑
2015/05/05
0
0

没有更多内容

加载失败,请刷新页面

加载更多

深入理解OAuth2.0协议

1. 引言 如果你开车去酒店赴宴,你经常会苦于找不到停车位而耽误很多时间。是否有好办法可以避免这个问题呢?有的,听说有一些豪车的车主就不担心这个问题。豪车一般配备两种钥匙:主钥匙和泊...

xtof
25分钟前
1
0
Linux学习-0920

3.4 usermod命令 3.5 用户密码管理 3.6 mkpasswd命令 一、usermode命令 usermode作用是用来修改用户信息。 方法: usermod 参数 username 示例1:修改用户uid usermod -u 1010 test5 示例2...

wxy丶
35分钟前
1
0
synchronized锁对象的坑

今天本来写点其他东西,碰巧写了一下synchronized,没想到掉坑里面了,大佬别笑。 起初代码大概是这样的: package com.ripplechan.part_1_2_3;import java.util.concurrent.CountDownL...

RippleChan
38分钟前
1
0
XAMPP环境搭建(Apache + MariaDB + PHP + Perl)

operation system:ubuntu-18.04.1 step1:download XAMPP #sudo wget https://www.apachefriends.org/xampp-files/7.2.9/xampp-linux-x64-7.2.9-0-installer.run step2:install XAMPP #sudo ......

硅谷课堂
40分钟前
0
0
关于获取3DS MAX中的蒙皮数据 3DSMAX C++API的应用

目的是为OSG做自定义的导出插件. 记录取得数据的方法. Max在代码中会提供一个INode对象. 从这个对象里取出各种数据. getSkin这个函数取出了ISkin修改器 下面这个函数一样是用于学习, 把数据输...

洛克人杰洛
47分钟前
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部