KafkaSpout（idle）产生了巨大的网络流量

Question

在使用KafkaSpout和几个Bolt开发并执行了我的Storm（1.0.1）拓扑之后，我注意到即使拓扑处于空闲状态，也会出现巨大的网络流量（Kafka上没有消息，Bolt中没有处理）。因此，我开始逐一评论我的拓扑结构，以便找到原因，现在我的主要内容中只有Kafka普特：

....
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，启动它并创建了一个简单的主题。当我在主题中放置消息时，拓扑中的所有内容都正常工作，没有任何问题）

显然，原因就在《Kafka普特》中。nextTuple（）method（如下），但我不明白，如果没有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

你的螺栓收到信息了吗？你的螺栓继承BaseRichBolt吗？

注释掉Kafaspout中的m.fail（id.offset）行并检查它。如果您的插销没有确认，那么您的喷口将假定消息失败，并尝试重播相同的消息。

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

也尝试停止nextTuple（）几毫升，并检查出来。

如果有帮助就告诉我

Answer 2

例如，在nextTuple（）方法中Hibernate一秒钟（1000毫秒），然后立即观察流量，

@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消费者基于拉动方法工作，这意味着消费者将从Kafka经纪人那里拉动数据。因此，从消费者的角度来看（Kafka Spoot）将持续向Kafka代理发出一个获取请求，这是一个TCP网络请求。因此，您面临着发送/接收数据包的大量统计数据<代码>虽然使用者不使用任何消息，但拉取请求和空响应也将计入网络数据包发送/接收统计数据中