当应用程序在pcf中自动缩放时,kafka应答超时

ymdaylpp  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(412)

我使用replyingkafkatemplate来实现kafka同步响应,并且我能够在只有一个示例运行时获得响应。但是如果应用程序扩展到多个示例,我就会得到超时错误。
来自文档
配置单个回复主题时,每个示例必须使用不同的group.id。在这种情况下,所有示例都会收到每个回复。
根据文档,如果我们需要使用不同的使用者组,这是否意味着我们需要手动运行具有不同使用者组的示例?如果我们使用pcf之类的工具,我们如何处理自动缩放。下面是我的Kafka配置。

@Configuration
@EnableKafka
public class KafkaConfig {

    //My Properties

    @Bean
    public Map < String, Object > producerConfig() {
        Map < String, Object > props = new HashMap < > ();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    @Bean
    public Map < String, Object > consumerConfig() {
        Map < String, Object > props = new HashMap < > ();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, commitInterval);
        return props;
    }

    @Bean
    public ProducerFactory < String, String > producerFactory() {
        return new DefaultKafkaProducerFactory(producerConfig());
    }

    @Bean
    public ConsumerFactory < String, String > consumerFactory() {
        return new DefaultKafkaConsumerFactory(consumerConfig());
    }

    @Bean
    public KafkaTemplate < String, String > kafkaTemplate() {
        return new KafkaTemplate(producerFactory());
    }

    @Bean
    public ReplyingKafkaTemplate < String, String, String > replyingKafkaTemplate
                            (ProducerFactory < String, String > pf, KafkaMessageListenerContainer < String, String > container) {
        ReplyingKafkaTemplate < String, String, String > rkt = new ReplyingKafkaTemplate(pf, container);
        rkt.setDefaultReplyTimeout(Duration.ofMillis(slaTime));
        rkt.setSharedReplyTopic(true);
        return rkt;
    }

    @Bean
    public KafkaMessageListenerContainer < String, String > replyContainer(ConsumerFactory < String, String > cf) {
        ContainerProperties containerProperties = new ContainerProperties(requestReplyTopic);
        containerProperties.setCommitLogLevel(LogIfLevelEnabled.Level.INFO);

        return new KafkaMessageListenerContainer < > (cf, containerProperties);
    }

    @Bean
    public KafkaListenerContainerFactory < ConcurrentMessageListenerContainer < String, String >> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory < String, String > factory = new ConcurrentKafkaListenerContainerFactory < > ();
        factory.setConsumerFactory(consumerFactory());
        factory.setReplyTemplate(replyingKafkaTemplate(producerFactory(), replyContainer(consumerFactory())));
        return factory;
    }
}
nwlls2ji

nwlls2ji1#

replyContainer 豆子,加

containerProperties.setGroupId(UUID.randomUUID().toString()); // unique
Properties props = new Properties();
props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // so the new group doesn't get old replies
containerProperties.setKafkaConsumerProperties(props);

replyingKafkaTemplate ,添加

rkt.setSharedReplyTopic(true);

请求主题至少需要与最大扩展一样多的分区。回复主题可以有任意数量的分区(包括1个分区)。
使用pcf,可以使用 instanceIndex 而不是随机的。
你也可以用 instanceIndex 作为 REPLY_PARTITION 报头和使用固定的应答分区;在这种情况下,您至少需要与最大分区数相同的分区数 instanceIndex 您希望使用。

相关问题