本文主要讲了笔者使用Strom中的一些优化建议

小鱼儿 2021-10-18 09:54 266阅读 0赞

## 1、使用rebalance命令动态调整并发度 ##

Storm计算以topology为单位，topology提交到Storm集群中运行后，通过storm rebalance 命令可对topology进行动态调整。比如增加Topology的worker数，修改Bolt，Spout的并行执行数量 parallelism等，从而实现topology的动态调整，达到弹性计算的目的。（当然调整时要配合监控模块）

基本上主要有两种用法：  
1) storm rebalance topology-name -n new-work-num,  
调整指定topology的worknum。  
2)storm rebalance topology-name -e component=parallelism  
调整指定topology中指定component的并行数量.

注：Jstorm不提供这个功能

## 2、使用tick消息做定时器 ##

使用Storm组件的定时器需要为bolt重写下面的方法：  
public Map<String, Object> getComponentConfiguration() \{  
Map<String, Object> conf = new HashMap<String, Object>();  
conf.put(Config.TOPOLOGY\_TICK\_TUPLE\_FREQ\_SECS, 60);//每60s持久化一次数据  
return conf;  
\}  
其中：Config.TOPOLOGY\_TICK\_TUPLE\_FREQ\_SECS 定时消息发送的频率，单位为秒。  
我们判断是否为tick消息。可以使用TupleHelpers类中的isTickTuple方法，具体代码：  
public static boolean isTickTuple(Tuple tuple) \{  
return tuple.getSourceComponent().equals(Constants.SYSTEM\_COMPONENT\_ID) && tuple.getSourceStreamId().equals(  
Constants.SYSTEM\_TICK\_STREAM\_ID);  
\}

## 3、使用组件的并行度代替线程池 ##

道理很简单，通过给bolt/spout 设置parallelism，而不是在bolt/spout设置线程池。这样可以避免一个组件将一台机器资源耗尽。

## 4、不要在Spout中处理耗时操作 ##

Spout都是单线程的，如果太多的耗时操作就在这里，那么整个程序的处理吞吐量就会下降。而且如果nextTuple方法非常耗时，那么消息会成功执行完毕后，Acker给Spout发送消息，Spout如果无法及时消费，可能造成Ack消息超时丢弃。然后Spout就认为这个消息执行失败

## 5、单个spout/bolt中做一件事 ##

每个spout/bolt只做一件事，比如消息的第一次清洗放在spout,接下来的计算放在bolt，再接下来计算结果入库再放到一个bolt，将整个任务串连多个spout/bolt起来。而不是单独的放在一个组件中。这样才能最大使用集群的资源。也好方便对一个单独的组件进行调优

## 6、注意fieldsGrouping的数据均衡性 ##

Strom中有6种fields方法，fieldsGrouping是按字段进行分组.通过合理的设置。确实各个Bolt/spout接收的消息都比较均衡。避免单个节点机器处理大量数据，这样耗时又耗机器。这里就涉及到一个field字段的选择。

## 7、优先使用LocalOrShuffleGrouping ##

数据首先优先选择本节点上的bolt处理，减少不必要的数据传输。其它Storm Grouping如下，

Shuffle Grouping ：随机分组，尽量均匀分布到下游Bolt中  
将流分组定义为混排。这种混排分组意味着来自Spout的输入将混排，或随机分发给此Bolt中的任务。shuffle grouping对各个task的tuple分配的比较均匀。  
Fields Grouping ：按字段分组，按数据中field值进行分组；相同field值的Tuple被发送到相同的Task  
这种grouping机制保证相同field值的tuple会去同一个task，这对于WordCount来说非常关键，如果同一个单词不去同一个task，那么统计出来的单词次数就不对了。“if the stream is grouped by the “user-id” field, tuples with the same “user-id” will always go to the sae task”  
All grouping ：广播  
广播发送， 对于每一个tuple将会复制到每一个bolt中处理。  
Global grouping ：全局分组，Tuple被分配到一个Bolt中的一个Task，实现事务性的Topology。  
Stream中的所有的tuple都会发送给同一个bolt任务处理，所有的tuple将会发送给拥有最小task\_id的bolt任务处理。  
None grouping ：不分组  
不关注并行处理负载均衡策略时使用该方式，目前等同于shuffle grouping,另外storm将会把bolt任务和他的上游提供数据的任务安排在同一个线程下。  
Direct grouping ：直接分组 指定分组  
由tuple的发射单元直接决定tuple将发射给那个bolt，一般情况下是由接收tuple的bolt决定接收哪个bolt发射的Tuple。这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。 只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的taskid (OutputCollector.emit方法也会返回taskid)。

## 8、不是所有场景都要使用ACK机制 ##

如果说处理的消息每天都有上亿，消息丢失个几百几千其实对计算结果的影响是非常小的。先说一下ACK机制：

为了保证数据能正确的被处理, 对于spout产生的每一个tuple, storm都会进行跟踪。这里面涉及到ack/fail的处理，如果一个tuple处理成功是指这个Tuple以及这个Tuple产生的所有Tuple都被成功处理, 会调用spout的ack方法；如果失败是指这个Tuple或这个Tuple产生的所有Tuple中的某一个tuple处理失败, 则会调用spout的fail方法；  
在处理tuple的每一个bolt都会通过OutputCollector来告知storm, 当前bolt处理是否成功。  
另外需要注意的，当spout触发fail动作时，不会自动重发失败的tuple，需要我们在spout中重新获取发送失败数据，手动重新再发送一次。  
**Ack原理**

Storm中有个特殊的task名叫acker，他们负责跟踪spout发出的每一个Tuple的Tuple树（因为一个tuple通过spout发出了，经过每一个bolt处理后，会生成一个新的tuple发送出去）。当acker（框架自启动的task）发现一个Tuple树已经处理完成了，它会发送一个消息给产生这个Tuple的那个task。  
Acker的跟踪算法是Storm的主要突破之一，对任意大的一个Tuple树，它只需要恒定的20字节就可以进行跟踪。  
Acker跟踪算法的原理：acker对于每个spout-tuple保存一个ack-val的校验值，它的初始值是0，然后每发射一个Tuple或Ack一个Tuple时，这个Tuple的id就要跟这个校验值异或一下，并且把得到的值更新为ack-val的新值。那么假设每个发射出去的Tuple都被ack了，那么最后ack-val的值就一定是0。Acker就根据ack-val是否为0来判断是否完全处理，如果为0则认为已完全处理。

**使用IBasicBolt接口实现自动确认**

为了简化编码，Storm为Bolt提供了一个IBasicBolt接口，它会在调用execute()方法之后正确调用ack()方法，BaseBasicBolt类是该接口的一个实现，用来实现自动确认

## 9、尽量抽取公用的处理部分到一个组件 ##

比如说存储到数据库的动作。可以尽量都使用同个bolt来写，管理好线程池

## 10、合理设置work数目 ##

work数目并不是越多越好，还要看你的数据和你的处理逻辑。这个一般情况下可以反复调整参数来确定最优，可以通过查看页面上的各个bolt/spout处理消息的耗时，资源使用情况来确定

## 11、GC参数优化 ##

对每个work的jvm参数进行调整，推荐生产使用parNew+CMS垃圾回收方式。

上面就是笔者自己的一些总结，希望对你有帮助。