0

我正在使用ReplyingKafkaTemplate进行 Kafka 同步响应，并且仅在实例运行一次时才能获得响应。但是，如果应用程序扩展到多个实例，我会收到超时错误。

从文档

配置单个回复主题时，每个实例必须使用不同的 group.id。在这种情况下，所有实例都会收到每个回复。

根据文档，如果我们需要使用不同的消费者组，这是否意味着我们需要手动运行具有不同消费者组的实例？如果我们使用 PCF 之类的工具，我们如何处理自动缩放。下面是我的卡夫卡配置。

@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;
    }
}
4

1 回答 1

1

在replyContainerbean 中，添加

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，您可以使用 构造 groupIdinstanceIndex而不是使其随机化。

您也可以将instanceIndex用作REPLY_PARTITION标头并使用固定回复分区；在这种情况下，您至少需要与instanceIndex您期望使用的最大值一样多的分区。

于 2020-08-26T20:12:16.030 回答