storm源码分析研究（十一）
作者：互联网
2021SC@SDUSC
Trident的Bolt节点分析
2021SC@SDUSC
SubTopologyBolt类型为Trident中运行的基本单位，但它并不是真正的Bolt节点，Trident会利用TridentBoltExecutor对SubTopologyBolt进行接口适配。
TridentBoltExecutor继承自IRichBolt接口，是Trident中真正运行的Bolt节点。它提供了类似于协调Bolt ( CoordinatedBolt )节点的功能，通过发送协调消息来对各个节点进行同步。
SubTopologyBolt主要用于对TridentProcessor的执行进行抽象。本篇文章将讨论SubTopologyBolt和TridentBoltExecutor的实现。
SubtopologyBolt.java
类的定义为：
public class SubtopologyBolt implements ITridentBatchBolt {
private static final long serialVersionUID = 1475508603138688412L;
@SuppressWarnings("rawtypes")
final DirectedGraph<Node, IndexedEdge> graph;
final Set<Node> nodes;
final Map<String, InitialReceiver> roots = new HashMap<>();
final Map<Node, Factory> outputFactories = new HashMap<>();
final Map<String, List<TridentProcessor>> myTopologicallyOrdered = new HashMap<>();
final Map<Node, String> batchGroups;
graph：整个Topology所对应的有向图。
nodes：该Bolt中所包含的处理节点。_nodesS_graph节点的子集。
roots：每种类型的输人流都会对应一个lnitalReceiver对象，用于表示如何处理该流的消息。
outputFactories：每个处理节点都会对应一个输出的工厂。 myTopologicallyOrdered：它的键为节点组序号，值为节点组所对应的TridentProcessor。
batchGroups：该变量保存了反向的索引，用来表示每个节点属于哪一个节点组。
BatchGroup对应于graph中的一个最大连通子图。
主要方法：
public void prepare(Map<String, Object> conf, TopologyContext context, BatchOutputCollector batchCollector) {
int thisComponentNumTasks = context.getComponentTasks(context.getThisComponentId()).size();
for (Node n : nodes) {
if (n.stateInfo != null) {
State s = n.stateInfo.spec.stateFactory.makeState(conf, context, context.getThisTaskIndex(), thisComponentNumTasks);
context.setTaskData(n.stateInfo.id, s);
}
}
DirectedSubgraph<Node, ?> subgraph = new DirectedSubgraph<>(graph, nodes, null);
TopologicalOrderIterator<Node, ?> it = new TopologicalOrderIterator<>(subgraph);
int stateIndex = 0;
while (it.hasNext()) {
Node n = it.next();
if (n instanceof ProcessorNode) {
ProcessorNode pn = (ProcessorNode) n;
String batchGroup = batchGroups.get(n);
if (!myTopologicallyOrdered.containsKey(batchGroup)) {
myTopologicallyOrdered.put(batchGroup, new ArrayList<>());
}
myTopologicallyOrdered.get(batchGroup).add(pn.processor);
List<String> parentStreams = new ArrayList<>();
List<Factory> parentFactories = new ArrayList<>();
for (Node p : TridentUtils.getParents(graph, n)) {
parentStreams.add(p.streamId);
if (nodes.contains(p)) {
parentFactories.add(outputFactories.get(p));
} else {
if (!roots.containsKey(p.streamId)) {
roots.put(p.streamId, new InitialReceiver(p.streamId, getSourceOutputFields(context, p.streamId)));
}
roots.get(p.streamId).addReceiver(pn.processor);
parentFactories.add(roots.get(p.streamId).getOutputFactory());
}
}
List<TupleReceiver> targets = new ArrayList<>();
boolean outgoingNode = false;
for (Node cn : TridentUtils.getChildren(graph, n)) {
if (nodes.contains(cn)) {
targets.add(((ProcessorNode) cn).processor);
} else {
outgoingNode = true;
}
}
if (outgoingNode) {
targets.add(new BridgeReceiver(batchCollector));
}
TridentContext triContext = new TridentContext(
pn.selfOutFields,
parentFactories,
parentStreams,
targets,
pn.streamId,
stateIndex,
batchCollector
);
pn.processor.prepare(conf, context, triContext);
outputFactories.put(n, pn.processor.getOutputFactory());
}
stateIndex++;
}
}
for (Node n : nodes)：
在此循环中判断节点的statelnfo是否为空，并对存储的State对象进行初始化。初始化过后的State对象被存储于TopologyContext的taskData中，并以statelnfo.id为键。statelnfo.id是一个以串 “ state” 为前缀的在Topology中唯一的字符串。
subgraph：
根据SubTopologyBolt中包含的节点获得一个子图。
it：
对子图进行拓扑排序，TopologicalOrderlterator类型的it变量用来按照拓扑排序的顺序遍历子图。
if (n instanceof ProcessorNode)：
SubTopologyBolt只对处理节点进行操作。处理节点中包含了一个TridentProcessor。Spout节点和分区节点则不在SubTopologyBolt的处理范围之内。
pn.processor.prepare(conf, context, triContext：
调用该TridentProcessor的prepare方法，它将新产生的TridentContext对象作为构造函数的一个参数传入。
outputFactories.put(n, pn.processor.getOutputFactory())：
将该TridentProcessor所对应的输出添加到SubTopologyBolt的输出中。于是该输出便可被其他的SubTopologyBolt作为输入了。
statelndex变量：
用于唯一地标识SubTopologyBolt中的每一个节点。
public void execute(BatchInfo batchInfo, Tuple tuple) {
String sourceStream = tuple.getSourceStreamId();
InitialReceiver ir = roots.get(sourceStream);
if (ir == null) {
throw new RuntimeException("Received unexpected tuple " + tuple.toString());
}
ir.receive((ProcessorContext) batchInfo.state, tuple);
}
首先，根据输人消息的流号，在roots中找到对应的InitialReceiver对象，并调用其receive方法。
所有等待该流消息的TridentProcessor的execute方法均会被调用。
在某个TridentProcessor的execute方法中，下游TridentProcessor的execute方法也会被依次调用到，于是构成了一个调用链，直到SubTopologyBolt完成对该消息的处理后结束。
public void finishBatch(BatchInfo batchInfo) {
for (TridentProcessor p : myTopologicallyOrdered.get(batchInfo.batchGroup)) {
p.finishBatch((ProcessorContext) batchInfo.state);
}
}
@Override
public Object initBatchState(String batchGroup, Object batchId) {
ProcessorContext ret = new ProcessorContext(batchId, new Object[nodes.size()]);
for (TridentProcessor p : myTopologicallyOrdered.get(batchGroup)) {
p.startBatch(ret);
}
return ret;
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
for (Node n : nodes) {
declarer.declareStream(n.streamId, TridentUtils.fieldsConcat(new Fields("$batchId"), n.allOutputFields));
}
}
在initBatchState方法中，会对ProcessorContext的数据进行初始化，然后返回ProcessorContext对象。Trident中的聚集器均要基于ProcessorContext中state所存储的数据来实现。
declareOutputFields方法会对SubTopologyBolt中每一个节点的输出进行声明，它将$batchid作为第1列。可以看出SubTopologyBolt虽然作为一个整体而存在，可其中每一个节点的输出均可能成为最终的输出。
标签：十一,context,TridentProcessor,源码,storm,new,nodes,节点,SubTopologyBolt 来源： https://blog.csdn.net/qq_45849855/article/details/121716312