卡夫卡喷口性能不佳

Question

我正在测试简单的拓扑来检查 kafka spout 的性能。 它包含确认每个元组的 kafka spout 和 Bolt。 螺栓执行方法：

public void execute(Tuple input) {
    collector.ack(input);
}

拓扑如下所示：

protected void configureTopology(TopologyBuilder topologyBuilder) {
    configureKafkaCDRSpout(topologyBuilder);
    configureKafkaSpoutBandwidthTesterBolt(topologyBuilder);
}

private void configureKafkaCDRSpout(TopologyBuilder builder) {
    KafkaSpout kafkaSpout = new KafkaSpout(createKafkaCDRSpoutConfig());
    int spoutCount = Integer.valueOf(topologyConfig.getProperty("kafka.cboss.cdr.spout.thread.count"));
    builder.setSpout(KAFKA_CDR_SPOUT_ID, kafkaSpout, spoutCount)
            .setNumTasks(Integer.valueOf(topologyConfig.getProperty(KAFKA_CDR_SPOUT_NUM_TASKS)));
}
private SpoutConfig createKafkaCDRSpoutConfig() {
    BrokerHosts hosts = new ZkHosts(topologyConfig.getProperty("kafka.zookeeper.broker.host"));
    String topic = topologyConfig.getProperty("kafka.cboss.cdr.topic");
    String zkRoot = topologyConfig.getProperty("kafka.cboss.cdr.zkRoot");
    String consumerGroupId = topologyConfig.getProperty("kafka.cboss.cdr.consumerId");
    SpoutConfig kafkaSpoutConfig = new SpoutConfig(hosts, topic, zkRoot, consumerGroupId);
    kafkaSpoutConfig.scheme = new SchemeAsMultiScheme(new CbossCdrScheme());
    kafkaSpoutConfig.ignoreZkOffsets = true;
    kafkaSpoutConfig.fetchSizeBytes = Integer.valueOf(topologyConfig.getProperty("kafka.fetchSizeBytes"));
    kafkaSpoutConfig.bufferSizeBytes = Integer.valueOf(topologyConfig.getProperty("kafka.bufferSizeBytes"));
    return kafkaSpoutConfig;
}

public void configureKafkaSpoutBandwidthTesterBolt(TopologyBuilder topologyBuilder) {
    SimpleAckerBolt b = new SimpleAckerBolt();
    topologyBuilder.setBolt(SPOUT_BANDWIDTH_TESTER_BOLT_ID, b, Integer.valueOf(topologyConfig.getProperty(CFG_SIMPLE_ACKER_BOLT_PARALLELISM)))
            .setNumTasks(Integer.valueOf(topologyConfig.getProperty(SPOUT_BANDWIDTH_TESTER_BOLT_NUM_TASKS)))
            .localOrShuffleGrouping(KAFKA_CDR_SPOUT_ID);
}

其他拓扑设置：

topology.max.spout.pending=250
topology.executor.receive.buffer.size=1024
topology.executor.send.buffer.size=1024
topology.receiver.buffer.size=8
topology.transfer.buffer.size=1024
topology.acker.executors=1

我正在使用 1 个工人、1 个 Kafka Spout 和 1 个 Simple Acker Bolt 启动我的拓扑。 这就是我在风暴 UI 中得到的：

好吧，我在 10 分钟内得到了 1.5kk 个元组。螺栓容量约为 0,5。所以我的逻辑很简单：如果我双喷嘴和螺栓并行提示 - 我将获得双倍性能。 下一个测试是使用 1 个 worker 2 个 Kafka Spout、2 个 Simple Acker Bolt 和 topology.acker.executors=2。结果如下：

因此，随着并行提示的增加，我的性能会变得更差。为什么会发生？如何增加每秒处理的元组？实际上，任何 spout 并行提示大于 2 的测试都显示出比 1 spout executor 更差的结果。

我已经检查过了：
1）这不是卡夫卡的错。主题在 2 个代理上有 20 个分区。 4 个工人规模的拓扑并获得 x4 性能。
2）这不是服务器故障。服务器有 40 个内核和 32Gb RAM。在运行拓扑时，它消耗大约 1/8 的 CPU 和几乎没有 RAM。
3) 更改 topology.max.spout.pending 参数没有帮助。
4) 增加 Bolt 或 Acker 并行性提示也无济于事。

Answer 1

因此，您的一名工人的表现似乎达到了极限。你只是让一个工人去做很多事情，它无法处理所有事情。

此时，如果您想进一步提高系统性能，您有两种选择。

1. 添加更多工作人员。
2. 提高一个工人的工作能力。

如果您不想添加更多工人，那么剩下的就是配置您的一名工人。然后，您应该调查您的一名工作人员的配置，为其提供更多内存、更多 cpu 等。您可能应该查看 Storm 的 default configuration options 并查看调整某些配置值是否会给您带来更好的性能。一些似乎比其他配置更有帮助的配置：

worker.heap.memory.mb:
worker.childopts:
supervisor.childopts:
supervisor.memory.capacity.mb:
supervisor.cpu.capacity: