Как распараллелить потребление сообщений Kafka

Question

У меня есть такой код:

@Component
public class PostConsumer {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(KafkaProperties kafkaProperties) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =           new ConcurrentKafkaListenerContainerFactory<>();
        ...
        factory.setConcurrency(10);    
        return factory;
    }

    @Autowired
    private PostService postService;

    @KafkaListener(topics = {"topic1", "topic2"} , containerFactory = "postKafkaListenerContainerFactory")
    public void sendPost(ConsumerRecord<String, Post> consumerRecord) {

        postService.sendPost(consumerRecord.value());

    }
}

Как выполнить параллельную обработку раздела (ов) topic1 и раздела (ов) topic2?

Сейчас они работают по одномуодин, сначала обработка topic1-> partition0, затем topic2-> partition0, затем снова topic1-> partition0 и т. д.

PS Когда я разделяю этот метод на два метода @KafkaListener, он утверждает, что работает параллельно, но проблема заключается в том, чтов списке тем я предоставляю огромный список тем, не хочу, чтобы для этого было более 100+ методов @KafkaListener.

Заранее спасибо.

Answer 1

Вам необходимо добавить параллелизм в контейнер слушателя. Однако, если каждая тема имеет только один раздел, это не поможет; вам нужно достаточно разделов в каждой теме, чтобы поддерживать желаемый параллелизм, или использовать отдельный контейнер для каждой темы.

Мы могли бы рассмотреть альтернативную стратегию с возможностью распределения самих тем среди потребителей.

Откройте вопрос GitHub , и мы его рассмотрим.

EDIT

Вы также можете изменить присваиватель разделов по умолчанию, например:

spring.kafka.consumer.properties.partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

и

@SpringBootApplication
public class Kgh941Application {

    public static void main(String[] args) {
        SpringApplication.run(Kgh941Application.class, args);
    }

    @KafkaListener(id = "kgh941", concurrency = "30",
            topics = {"kgh941a", "kgh941b", "kgh941c", "kgh941d", "kgh941e", "kgh941f", "kgh941g", "kgh941h"})
    public void listen(String in) {

    }

    @Bean
    public ApplicationRunner runner(KafkaListenerEndpointRegistry registry) {
        return args -> {
            System.out.println("Hit enter to get assignments");
            System.in.read();
            MessageListenerContainer container = registry.getListenerContainer("kgh941");
            @SuppressWarnings("unchecked")
            List<KafkaMessageListenerContainer<?, ?>> containers = (List<KafkaMessageListenerContainer<?, ?>>) new DirectFieldAccessor(
                    container).getPropertyValue("containers");
            containers.forEach(c -> System.out.println(c.getAssignedPartitions().size() + ":"
                    + c.getAssignedPartitions()));
        };
    }

    @Bean
    public NewTopic topica() {
        return new NewTopic("kgh941a", 10, (short) 1);
    }

    @Bean
    public NewTopic topicb() {
        return new NewTopic("kgh941b", 10, (short) 1);
    }

    @Bean
    public NewTopic topicc() {
        return new NewTopic("kgh941c", 10, (short) 1);
    }

    @Bean
    public NewTopic topicd() {
        return new NewTopic("kgh941d", 10, (short) 1);
    }

    @Bean
    public NewTopic topice() {
        return new NewTopic("kgh941e", 10, (short) 1);
    }

    @Bean
    public NewTopic topicf() {
        return new NewTopic("kgh941f", 10, (short) 1);
    }

    @Bean
    public NewTopic topicg() {
        return new NewTopic("kgh941g", 10, (short) 1);
    }

    @Bean
    public NewTopic topich() {
        return new NewTopic("kgh941h", 10, (short) 1);
    }

}

и

3:[kgh941b-0, kgh941h-0, kgh941e-0]
2:[kgh941c-1, kgh941f-1]
2:[kgh941c-4, kgh941f-4]
2:[kgh941c-5, kgh941f-5]
2:[kgh941f-6, kgh941c-6]
2:[kgh941c-7, kgh941f-7]
2:[kgh941c-8, kgh941f-8]
2:[kgh941c-9, kgh941f-9]
3:[kgh941d-0, kgh941a-0, kgh941g-0]
3:[kgh941a-1, kgh941d-1, kgh941g-1]
3:[kgh941a-2, kgh941d-2, kgh941g-2]
3:[kgh941a-3, kgh941g-3, kgh941d-3]
3:[kgh941d-4, kgh941a-4, kgh941g-4]
3:[kgh941a-5, kgh941d-5, kgh941g-5]
3:[kgh941a-6, kgh941d-6, kgh941g-6]
3:[kgh941a-7, kgh941g-7, kgh941d-7]
3:[kgh941d-8, kgh941a-8, kgh941g-8]
3:[kgh941d-9, kgh941g-9, kgh941a-9]
3:[kgh941e-1, kgh941b-1, kgh941h-1]
3:[kgh941b-2, kgh941e-2, kgh941h-2]
3:[kgh941b-3, kgh941e-3, kgh941h-3]
3:[kgh941b-4, kgh941h-4, kgh941e-4]
3:[kgh941e-5, kgh941b-5, kgh941h-5]
3:[kgh941b-6, kgh941e-6, kgh941h-6]
3:[kgh941b-7, kgh941e-7, kgh941h-7]
3:[kgh941b-8, kgh941h-8, kgh941e-8]
3:[kgh941e-9, kgh941b-9, kgh941h-9]
2:[kgh941c-0, kgh941f-0]
2:[kgh941f-2, kgh941c-2]
2:[kgh941c-3, kgh941f-3]