为多个实体配置ReplyingKafkaTemplate

Question

我有2个服务（都是在kotlin上的Spring引导）。将其命名为“客户机”和“服务器”，由于某些限制，我必须使用Kafka同步请求-应答模式。所以我尝试使用ReplyingKafkatemplate。我的问题是我需要为多个实体使用。意味着创建多个ReplyingKafkaTemplate，一个用于“foo”，第二个用于“bar”。因此，在我的代码中，我用setup创建了多个KafkaConfig类，每个实体和一个基本配置。

另外，我将KafkaAutoConfig排除在加载之外。下面是“服务器”端的配置（W/O kreplying kafka模板）：

@Configuration
@EnableKafka
class KafkaConfig @Autowired constructor(
    @Value("\${kafka.bootstrap-servers}")
    private var bootstrapServers: String,
    @Value("\${kafka.consumer-group.name}")
    private var consumerGroup: String,
    @Value("\${kafka.consumer-group.id}")
    private var groupId: Number
) {

    @Bean("kafkaProducerConfig")
    fun producerConfigs(): MutableMap<String, Any> {
        return mutableMapOf(
            Pair(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9093,kafka3:9094"),
            Pair(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java),
            Pair(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer::class.java),
            Pair(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"),
            Pair(ProducerConfig.ACKS_CONFIG, "all"),
            Pair(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5"),
            Pair(ProducerConfig.RETRIES_CONFIG, Int.MAX_VALUE.toString()),
            Pair(ProducerConfig.LINGER_MS_CONFIG, "20"),
            Pair(ProducerConfig.BATCH_SIZE_CONFIG, (32 * 1024).toString()),
            Pair(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy")
        )
    }

    @Bean("kafkaConsumerConfig")
    fun consumerConfigs(): Map<String, Any> {
        return mutableMapOf(
            Pair(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9093,kafka3:9094"),
            Pair(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer::class.java),
            Pair(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonSerializer::class.java),
            Pair(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"),
            Pair(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup + groupId)
        )
    }

    @Bean("kafkaAdminConfig")
    fun admin(): KafkaAdmin {
        val configs: MutableMap<String, Any> = HashMap()
        configs[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = bootstrapServers
        return KafkaAdmin(configs)
    }
}

FOO（BAR与FOO相同的配置，而不是其他实体和bean的命名）配置是：

@Configuration
class KafkaConfigForFOO {

    @Value("\${kafka.topic.request-consumable-topic}")
    private lateinit var requestConsumableTopic: String

    @Value("\${kafka.request-reply.timeout-ms}")
    private lateinit var replyTimeout: Number

    @Bean("requestFOOTopicConfig")
    fun requestConsumableTopic(): NewTopic {
        val configs: MutableMap<String, String> = HashMap()
        configs["retention.ms"] = replyTimeout.toString()
        return NewTopic(requestConsumableTopic, 6, 3.toShort()).configs(configs)
    }

    @Bean("producerFactoryForFOO")
    @Autowired
    fun producerFactoryForFOO(@Qualifier("kafkaProducerConfig") producerConfigs: MutableMap<String, Any>):
        ProducerFactory<String, FOO> = DefaultKafkaProducerFactory(producerConfigs)

    @Bean("kafkaTemplateForFOO")
    @Autowired
    fun kafkaTemplateForFOO(@Qualifier("producerFactoryForFOO") producerFactory: ProducerFactory<String, FOO>):
        KafkaTemplate<String, FOO> = KafkaTemplate(producerFactory)

    @Bean("consumerFactoryForFOO")
    @Autowired
    fun consumerFactoryForFOO(@Qualifier("kafkaConsumerConfig") consumerConfigs: MutableMap<String, Any>):
        ConsumerFactory<String, FOO> = DefaultKafkaConsumerFactory(consumerConfigs, StringDeserializer(), JsonDeserializer(FOO::class.java))

    @Bean("kafkaListenerContainerFactoryForFOO")
    @Autowired
    fun kafkaListenerContainerFactoryForFOO(
        @Qualifier("consumerFactoryForFOO") consumerFactory: ConsumerFactory<String, FOO>,
        @Qualifier("kafkaTemplateForFOO") kafkaTemplate: KafkaTemplate<String, FOO>
    ):
        KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, FOO>> {
        val factory = ConcurrentKafkaListenerContainerFactory<String, FOO>()
        factory.consumerFactory = consumerFactory
        factory.setReplyTemplate(kafkaTemplate)
        return factory
    }
}

@Component
class FOOReplyingKafkaConsumer @Autowired constructor(
    private val fooService: FooService
) {
    @KafkaListener(topics = ["\${kafka.topic.request-FOO-topic}"], containerFactory = "kafkaListenerContainerFactoryForFoo", groupId = "\${spring.kafka.consumer.group-id}")
    @SendTo()
    fun cropListen(request: FOO): FOO{
        return FOO(fooService.getAllByIds(request.ids ?: mutableSetOf()).toMutableSet())
    }
}

Answer 1

你不需要两个消费者工厂；类型擦除意味着它在运行时是不相关的。

引导将一个配置为

ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(

它实际上是 （或者在Kotlin中是 )。

public KafkaTemplate<?, ?> kafkaTemplate