我使用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;
}
}
1条答案
按热度按时间nwlls2ji1#
在
replyContainer
豆子,加在
replyingKafkaTemplate
,添加请求主题至少需要与最大扩展一样多的分区。回复主题可以有任意数量的分区(包括1个分区)。
使用pcf,可以使用
instanceIndex
而不是随机的。你也可以用
instanceIndex
作为REPLY_PARTITION
报头和使用固定的应答分区;在这种情况下,您至少需要与最大分区数相同的分区数instanceIndex
您希望使用。