如何在暴风雨中制作同步Kafka

Question

我试图使Kafka消费者同步消费Kafka的消息。

我遇到的实际问题是消息队列存储在Storm Spout中。

我想做的是让暴风雪等待Kafka的回复，然后让暴风雪消耗下一条信息。

我正在使用Storm KafkaSpout：

/**
     * Creates a configured kafka spout.
     * @param topicName Topic where the kafka spout subscribes
     * @return An instance of configured KafkaSpout
     */
    public KafkaSpout getkafkaSpout(String topicName){
        return new KafkaSpout(this.getSpoutConfig(topicName));
    }

    /**
     * Create the necessary configuration to create a new kafka spout.
     * @param topicName Topic where the kafka spout subscribes
     * @return Spout configuration
     */
    public SpoutConfig getSpoutConfig(String topicName) {
        SpoutConfig spoutConfig=new SpoutConfig(this.getZkHosts(),topicName, "", String.join("-",topicName,RandomStringUtils.randomAlphanumeric(20)));
        spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
        spoutConfig.startOffsetTime=kafka.api.OffsetRequest.LatestTime();
        return spoutConfig;
    }



builder.setSpout("kafkaPackedData", stormConfig.getkafkaSpout("topic_kafka"), 2);

我已经更新到Storm 2.0.0，我使用Storm kafka客户端。但是如果我将Storm队列配置为50:setMaxSpoutPending（50） 当我向Kafka发送许多数据时，它会停止使用这些数据。

我已经配置了Kafka消费者与下一个配置：

KafkaSpoutConfig spoutConf =  KafkaSpoutConfig.builder("stream1:9092", "kafkaToStormAlarms")
                    .setProp(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "50000") //Set session timeout
                    .setProp(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "60000") //Set request timeout
                    .setOffsetCommitPeriodMs(10000)    //Set automatic confirmation time (in ms)
                    .setFirstPollOffsetStrategy(LATEST)    //Set to pull the latest messages
                    .setRetry(kafkaSpoutRetryService)
                    .build();

当Storm使用与MaxSpoutPending配置相同的50条消息时，它将停止使用更多消息。也许下一个螺栓没有正确发送ACK？我使用Kafkaconsumerspoot之后的下一个螺栓：

public class testBolt extends BaseBasicBolt {

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("MQTTmessage"));
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector boc) {
        System.out.println("\n\n\n\nLLEGA BIENN AL SPLIT TEXT BOLT\n\n");
        System.out.println("TUPLE "+tuple);
        String text = tuple.getString(4);
        List<String> lines = Arrays.asList(text.split("\\r?\\n"));

        lines.forEach(line -> {
            boc.emit(new Values(line));
        });
    }
}




Answer 1

关于节流喷嘴：是的，可以通过设置拓扑来实现。马克斯，喷口。将拓扑配置中的挂起的选项设置为1。如果您想获得良好的吞吐量，我不推荐使用它，但我假设您已经仔细考虑了为什么需要拓扑以这种方式运行。

关于新的喷口：stream1:9092是Kafka正在运行的服务器吗，以及kafkaToStormAlarms是您要发送到的主题吗？如果不是，那可能是你的问题。否则，请在storm/logs/workers artifacts中检查工作日志，它可能会告诉您喷口没有排放任何东西的原因。

最后是的，您绝对应该使用Storm-kafka-Client而不是Storm-kafka，否则您将无法升级到Storm 2.0.0或最新的Kafka版本。