JStorm之Nimbus简介

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

@SuppressWarnings("rawtypes")
private void launchServer(Map conf) throws Exception {
LOG.info("Begin to start nimbus with conf " + conf);
//1.检查配置文件中是否配置为分布式模式
StormConfig.validate_distributed_mode(conf);
//2.注册主线程退出Hook现场清理（关闭线程+清理数据）
initShutdownHook();
//3.新建NimbusData数据，记录30s超时上传下载通道Channel/BufferFileInputStream
data = createNimbusData(conf);
//4.nimbus本地不存在的stormids数据如果在ZK上存在则删除，其中删除操作包括/zk/{assignments,tasks,storms}相关数据
NimbusUtils.cleanupCorruptTopologies(data);
//5.启动Topology分配策略
initTopologyAssign();
//6.初始化所有topology的状态为startup
initTopologyStatus();
//7.监控所有task的heartbeat,一旦发现taskid失去心跳将其置为needreassign 1次/10s
initMonitor(conf);
//8.启动cleaner线程，默认600s扫描一次，默认删除3600s没有读写过的jar文件
initCleaner(conf);
//9.初始化ServiceHandler
serviceHandler = new ServiceHandler(data);
//10.启动rpc server
initThrift(conf);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
2425
26
27
28
29
30
31
32
33
34

public static void submitTopology(
String name,
Map stormConf,
StormTopology topology)
throws AlreadyAliveException, InvalidTopologyException {
if(!Utils.isValidConf(stormConf)) {
throw new IllegalArgumentException("Storm conf is not valid.");
}
stormConf = new HashMap(stormConf);
stormConf.putAll(Utils.readCommandLineOpts());
Map conf = Utils.readStormConfig();
conf.putAll(stormConf);
try {
String serConf = JSONValue.toJSONString(stormConf);
if(localNimbus!=null) {
LOG.info("Submitting topology " + name + " in local mode");
localNimbus.submitTopology(name, null, serConf, topology);
} else {
//1.向Nimbus提交jar包
submitJar(conf);
NimbusClient client = NimbusClient.getConfiguredClient(conf);
try {
LOG.info("Submitting topology " +  name + " in distributed mode with conf "  + serConf);
//2.提交topology DAG及序列化后的配置信息serconf(json)
client.getClient().submitTopology(name, submittedJar, serConf, topology);
} finally {
client.close();
}
}
LOG.info("Finished submitting topology: " +  name);
} catch(TException e) {
throw new RuntimeException(e);
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
2425
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

@SuppressWarnings("unchecked")
@Override
public void submitTopology(String topologyname, String uploadedJarLocation, String jsonConf, StormTopology topology)
throws AlreadyAliveException, InvalidTopologyException, TException {
……
try {
//1.检测topologyName是否已经存在，如果存在相同名称的topology则拒绝提交
checkTopologyActive(data, topologyname, false);
}
……
//2.根据topologyname构造topologyid(=topologyname-$counter-$ctime)
int counter = data.getSubmittedCount().incrementAndGet();
String topologyId = topologyname + "-" + counter + "-"
+ TimeUtils.current_time_secs();
//3.根据输入参数jsonConf重组配置数据
Map serializedConf = (Map) JStormUtils.from_json(jsonConf);
if (serializedConf == null) {
throw new InvalidTopologyException("");
}
serializedConf.put(Config.STORM_ID, topologyId);
Map stormConf;
try {
stormConf = NimbusUtils.normalizeConf(conf, serializedConf, topology);
} catch (Exception e1) {
throw new TException(errMsg);
}
Map totalStormConf = new HashMap(conf);
totalStormConf.putAll(stormConf);
StormTopology newtopology = new StormTopology(topology);
//4.检查topology的合法性,包括componentid检查和spout/bolt不能为空检查
// this validates the structure of the topology
Common.validate_basic(newtopology, totalStormConf, topologyId);
try {
StormClusterState stormClusterState = data.getStormClusterState();
//5.在nimbus的本地准备所有topology相关数据
//包括$storm-local-dir/nimbus/stormdist/topologyid/{tormjar.jar,stormcode.ser,stormconf.ser}
// create $storm-local-dir/nimbus/topologyId/xxxx files
setupStormCode(conf, topologyId, uploadedJarLocation, stormConf,
newtopology);
//6.向zk写入task信息
//6.1新建目录$zkroot/taskbeats/topologyid
//6.2写文件$zkroot/tasks/topologyid/taskid 内容为对应task的taskinfo[内容：componentid]
// generate TaskInfo for every bolt or spout in ZK
// $zkroot /tasks/topoologyId/xxx
setupZkTaskInfo(conf, topologyId, stormClusterState);
//7.任务分配事件压入待分配队列
// make assignments for a topology
TopologyAssignEvent assignEvent = new TopologyAssignEvent();
assignEvent.setTopologyId(topologyId);
assignEvent.setScratch(false);
assignEvent.setTopologyName(topologyname);
TopologyAssign.push(assignEvent);
}
……
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
2425
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

public static List sortSlots(
Set allSlots, int needSlotNum) {
Map> nodeMap = new HashMap>();
// group by first
// 按照节点进行组织Map> : nodeid -> ports
for (NodePort np : allSlots) {
String node = np.getNode();
List list = nodeMap.get(node);
if (list == null) {
list = new ArrayList();
nodeMap.put(node, list);
}
list.add(np);
}

//每一个nodeid按照端口的大小进行排序
for (Entry> entry : nodeMap.entrySet()) {
List ports = entry.getValue();
Collections.sort(ports);
}

//收集所有的workers
List> splitup = new ArrayList>(nodeMap.values());

//按照节点可用worker数从小到大排序
//1.assignTasks-Map supInfos
//2.availSlots : splitup/List>
Collections.sort(splitup, new Comparator> () {
public int compare(List o1, List o2) {
return o1.size() - o2.size();
}
});

/*
* splitup目前的状态(A-F表示节点，1-h表示端口)
* |A| |B| |C| |D| |E| |F|
*--|---|---|---|---|---|--
* |1| |2| |3| |4| |5| |6|
*     |7| |8| |9| |0| |a|
*         |b| |c| |d| |e|
*                 |f| |g|
*                     |h|
* 经过interleave_all收集到的sortedFreeSlots为：
* 1-2-3-4-5-6-7-8-9-0-a-b-c-d-e-f-g-h
*/
List sortedFreeSlots = JStormUtils.interleave_all(splitup);

//比较sortedFreeSlots.size和needSlotNum的大小分配workers
if (sortedFreeSlots.size()  needSlotNum
return sortedFreeSlots.subList(0, needSlotNum);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
2425
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

@Override
public void run() {
//1.获取jstorm对zk的操作接口
StormClusterState clusterState = data.getStormClusterState();
try {
// Attetion, here don't check /ZK-dir/taskbeats to
// get active topology list
//2.通过$zkroot/assignments/获取所有需要检查active topology
List active_topologys = clusterState.assignments(null);

if (active_topologys == null) {
LOG.info("Failed to get active topologies");
return;
}

for (String topologyid : active_topologys) {
LOG.debug("Check tasks " + topologyid);
// Attention, here don't check /ZK-dir/taskbeats/topologyid to
// get task ids
//3.通过$zkroot/tasks/topologyid获取组成topology的所有tasks
List taskIds = clusterState.task_ids(topologyid);
if (taskIds == null) {
LOG.info("Failed to get task ids of " + topologyid);
continue;
}

boolean needReassign = false;
for (Integer task : taskIds) {
//4.检查task是否为Dead状态，主要是task心跳是否超时
boolean isTaskDead = NimbusUtils.isTaskDead(data, topologyid, task);
if (isTaskDead == true) {
needReassign = true;
break;
}
}
if (needReassign == true) {
//5.如果Topology里有Dead状态的Task则topology状态置为monitor等待任务分配线程重新分配
NimbusUtils.transition(data, topologyid, false, StatusType.monitor);
}
}
} catch (Exception e) {
// TODO Auto-generated catch block
LOG.error(e.getCause(), e);
}
}