风暴中的背压

Question

在我们的 Storm 1.0.2 应用程序中，我们面临内存不足的异常。在调试时，我们看到 Kafka spout 向 bolts 发出了太多消息。螺栓以几乎 4.0 的容量运行。那么有没有办法在暴风雨中启用背压，以便喷口根据螺栓的容量排放。尝试将 topology.backpressure.enable 启用为 true，但遇到了这个问题 https://issues.apache.org/jira/browse/STORM-1949。我们正在使用开箱即用的 KafkaSpout 实现并为我们的螺栓扩展 BaseRichBolt。我们的 DAG 是线性的。

Answer 1

您可以通过在拓扑配置中设置 ma​​xSpoutPending 值来处理 KafkaSpout 的背压，

Config config = new Config();
config.setMaxSpoutPending(200); 
config.setMessageTimeoutSecs(100);

StormSubmitter.submitTopology("testtopology", config, builder.createTopology());

maxSpoutPending 是在给定时间可以在拓扑中等待确认的元组数。设置此属性，将提示 KafkaSpout 不再使用来自 Kafka 的任何数据，除非未确认的元组计数小于 maxSpoutPending 值。