KafkaSpout（空闲）产生巨大的网络流量

Question

在使用 KafkaSpout 和几个 Bolt 开发和执行我的 Storm (1.0.1) 拓扑后，我注意到即使在拓扑空闲时也会产生巨大的网络流量（Kafka 上没有消息，bolts 中没有处理）。所以我开始逐条注释掉我的拓扑结构，以便找到原因，现在我的主目录中只有 KafkaSpout：

....
final SpoutConfig spoutConfig = new SpoutConfig(
                new ZkHosts(zkHosts, "/brokers"), 
                "files-topic", // topic
                "/kafka", // ZK chroot 
                "consumer-group-name");
     spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
     spoutConfig.startOffsetTime = OffsetRequest.LatestTime();
     topologyBuilder.setSpout(
                        "kafka-spout-id, 
                        new KafkaSpout(config),
                        1); 
....

当这个（无用的）拓扑执行时，即使是在本地模式下，即使是第一次，网络流量总是会增长很多：我看到（在我的活动监视器中）

• 平均每秒接收 432 KB 的数据
• 几个小时后，拓扑开始运行（空闲），接收到的数据为 1.26GB，发送的数据为 1GB

（重要提示：Kafka 没有在集群中运行，单个实例运行在具有单个主题和单个分区的同一台机器上。我刚刚在我的机器上下载了 Kafka，启动它并创建了一个简单的主题。当我把主题中的一条消息，拓扑中的所有内容都可以正常工作）

很明显，原因在KafkaSpout.nextTuple()方法（下），但我不明白为什么，在Kafka中没有任何消息，我应该有这样的流量。有什么我没有考虑到的吗？这是预期的行为吗？我看了看 Kafka 日志，ZK 日志，什么都没有，我已经清理了 Kafka 和 ZK 数据，什么都没有，仍然是相同的行为。

@Override
public void nextTuple() {
    List<PartitionManager> managers = _coordinator.getMyManagedPartitions();
    for (int i = 0; i < managers.size(); i++) {

        try {
            // in case the number of managers decreased
            _currPartitionIndex = _currPartitionIndex % managers.size();
            EmitState state = managers.get(_currPartitionIndex).next(_collector);
            if (state != EmitState.EMITTED_MORE_LEFT) {
                _currPartitionIndex = (_currPartitionIndex + 1) % managers.size();
            }
            if (state != EmitState.NO_EMITTED) {
                break;
            }
        } catch (FailedFetchException e) {
            LOG.warn("Fetch failed", e);
            _coordinator.refresh();
        }
    }

    long diffWithNow = System.currentTimeMillis() - _lastUpdateMs;

    /*
         As far as the System.currentTimeMillis() is dependent on System clock,
         additional check on negative value of diffWithNow in case of external changes.
     */
    if (diffWithNow > _spoutConfig.stateUpdateIntervalMs || diffWithNow < 0) {
        commit();
    }
}

Answer 1

在 nextTuple() 方法中休眠一秒（1000ms），现在观察流量，例如，

@Override
public void nextTuple() {
   try {
       Thread.sleep(1000);
   } catch(Exception ex){
        log.error("Ëxception while sleeping...",e);
   }
   List<PartitionManager> managers = _coordinator.getMyManagedPartitions();
   for (int i = 0; i < managers.size(); i++) {
     ...
     ...
     ...
     ...
}

---

原因是，kafka consumer 是在pull methodology 的基础上工作的，也就是说，consumer 会从 kafka broker 中拉取数据。因此，从消费者的角度来看（Kafka Spout）将不断地向 kafka 代理发出一个获取请求，即TCP network request。因此，您将面临发送/接收数据包的大量统计数据。 Though the consumer doesn't consumes any message, pull request and empty response also will get account into network data packet sent/received statistics. 如果您的睡眠时间较长，您的网络流量将会减少。还有一些network related configurations 用于经纪人和消费者。对配置进行研究可能会对您有所帮助。希望对你有帮助。

Answer 2

您的螺栓是否收到消息？您的螺栓是否继承了 BaseRichBolt ？

在 Kafaspout 中注释掉 m.fail(id.offset) 行并检查一下。如果你的 bolt 没有确认，那么你的 spout 会假设消息失败并尝试重播相同的消息。

public void fail(Object msgId) {
        KafkaMessageId id = (KafkaMessageId) msgId;
        PartitionManager m = _coordinator.getManager(id.partition);
        if (m != null) {
            //m.fail(id.offset);
        }

还可以尝试将 nextTuple() 暂停几毫秒并检查一下。

如果有帮助请告诉我