功能：提交一个新的Topology,并为Topology创建storm-id（topology-id）,校验其结构，设置必要的元数据，最后为Topology分配任务.
实现源码：
| 1 | ); Config conf = new Config(); |
| 4 | // 8对应 TOPOLOGY-MAX-TASK-PARALLELISM配置项的值 |
| 5 | Conf.setMaxTaskParallelism（8）; |
system-topology!
功能：
验证用户提交的Topology,同时为提交的topology添加一些系统组件和流。
实现源码：
| 1 | (defn system-topology! [storm-conf ^StormTopology topology] |
| 2 | (validate-basic! topology) |
| 3 | (let [ret (.deepCopy topology)] |
| 4 | (add-acker! storm-conf ret) |
| 5 | (add-metric-components! storm-conf ret) |
| 6 | (add-system-components! storm-conf ret) |
| 7 | (add-metric-streams! ret) |
| 8 | (add-system-streams! ret) |
| 9 | (validate-structure! ret) |
| 10 | ret |
| 11 | )) |
实现说明：
- 使用validate-basic!校验所提交的Topology.
主要用于确保topology中的组件id不重复而且不是系统id，以及确保每个组件的TOPOLOGY-TASKS配置项大于0时，组件的并行度设置也一定大于0. - 调用deepCopy对topology进行深度拷贝，赋值给ret.
- 为Topology添加acker-bolt.
用于追踪发送出去的消息是否被成功处理。 - 使用add-metric-components为Topology添加metric-bolt.
- 为Topology添加system-bolt.
System-bolt没有输入流只有输出流分别为：SYSTEM-TICK-STREAM-ID，声明字段是[“rate_secs”],非直接模式；另一个为METRICS-TICK-STREAM-ID,声明字段为[“interval”]非直接模式，并行度为0. - 为Topology中的所有组件添加统计流。
Stream-id为METRICS-STREAM-ID,声明字段为[“task-info”,”data-points”],非直接流模式. - 为Topology中的所有组件添加系统流。
stream-id为SYSTEM-STREAM-ID,声明字段为[“event”],非直接流模式. - 使用validate-structure!检验以上步骤所组合后的Topology.
验证过程：
获取Topology中所有组件和组件的输入(包括component-id、stream-id、Grouping),对输入组件依次判断输入组件ID(component-id)是否在该Topology中，不存在则抛出异常，存在则再判断该组件的流类型是否为所对应的stream-id，若不存在则抛出异常，存在则继续检查该流的分组方式（Grouping）是否与能对应，所有组件检查完毕后没有异常抛出表示该Topology有效.