Kafka吐槽糟糕的表演

Question

我正在测试简单拓扑以检查Kafka喷口的性能。它包含kafka spout和Bolt来确认每个元组。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

我在10分钟内得到了1.5kk元组。螺栓的性能约为0,5。所以我的逻辑很简单：如果我双倍喷注和螺栓并行提示-我将得到双倍的性能。下一个测试是1个工人2个Kafka喷口，2个简单的夹子螺栓和Topology.Acker.Executors=2。以下是结果：

所以，我得到了更差的性能增加并行化提示。为什么会发生？我如何增加元组每秒处理？实际上，任何喷口并行性提示大于2的测试都比1个喷口执行者显示出更差的结果。

我已经检查过:
1)这不是kafka错误。主题在2个Brokers上有20个分区。拓扑在4个workers上扩展并获得x4性能。
2)这不是服务器错误。服务器有40个内核和32GB RAM。运行拓扑时，它消耗了大约1/8的CPU和几乎没有RAM。
3)更改拓扑.max.spout.pending参数无济于事。
4)增加Bolt或Acker并行性提示也无济于事。

Answer 1

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

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

1. 添加更多员工。
2. 提高“一个员工”执行工作的能力。

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