Параллельные потоки прослушивателя Kafka требуют времени для параллельного запуска?

Мне нужно обработать около 50 тысяч записей (это число может варьироваться от 100 до 50 тысяч) в одной теме. Поэтому я использовал функцию параллелизма kafka. Ниже приведена моя конфигурация и код прослушивателя.

@KafkaListener(topics = {"kafkaTopic"},
                containerFactory = "abcd")
        public void consume(
                @Payload List<String> message,
                @Header(KafkaHeaders.RECEIVED_TOPIC) String topic
        ) throws IOException {
    
            StopWatch st = new StopWatch();
     DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
            Date date = new Date();
            StringBuilder str = new StringBuilder();
            st.start("threadName-");
            message.forEach(messages -> {
                try {
                    Thread.sleep(2500);
                    logger.info("message is-{}", messages);
                    str.append(messages);
                    str.append(",");
                } catch (Exception e) {
                    str.append("exception-{}" + e);
                }
            });
    
            st.stop();
           
            List data = objectMapper.readValue(getFile(), new TypeReference<List<String>>() {});
    
            str.append("----thread-" + Thread.currentThread().getName() + "started at time-"+dateFormat.format(date)+" and time taken-" + String.format("%.2f", st.getTotalTimeSeconds()));
            str.append("---");
            data.add(str);
            objectMapper.writeValue(getFile(),
                    data);
    
        }


    @Bean("abcd")
        public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
            factory.setConcurrency(5);
            factory.setBatchListener(true);
            return factory;
        }
    
        @Bean
        public NewTopic syliusDeTopic() {
            return TopicBuilder.name("kafkaTopic").partitions(5).replicas(2).build();
        }

@Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CustomCooperativeStickyAssignor.class.getName());
        configProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"500");
        configProps.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG,"1");
        configProps.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,"5000");
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

Но когда я проверил результат для образца 100 записей, потоки не запускались одновременно. Ниже приведен ответ на то же самое.

["test-0,test-1,test-2,----thread-org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1started at time-2023/01/06 22:20:19 and time taken-7.51---","test-56,test-57,test-58,test-59,test-60,test-61,----thread-org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1started at time-2023/01/06 22:20:26 and time taken-15.02---","test-70,test-71,test-72,test-73,test-74,test-75,test-76,test-77,test-78,----thread-org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1started at time-2023/01/06 22:20:34 and time taken-22.53---","test-62,test-63,test-64,test-65,test-66,test-67,test-68,test-69,test-85,test-86,test-87,test-88,test-89,test-90,test-91,----thread-org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1started at time-2023/01/06 22:20:49 and time taken-37.55---","test-79,test-80,test-81,test-82,test-83,test-84,test-92,test-93,test-94,test-95,test-96,test-97,test-98,test-99,----thread-org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1started at time-2023/01/06 22:21:01 and time taken-35.05---","test-3,test-4,test-5,test-6,test-7,test-8,test-9,test-10,test-11,test-12,test-13,test-14,test-15,test-16,test-17,test-18,test-19,test-20,test-21,test-22,test-23,test-24,test-25,test-26,test-27,test-28,test-29,test-30,test-31,test-32,test-33,test-34,test-35,test-36,test-37,test-38,test-39,test-40,test-41,test-42,test-43,test-44,test-45,test-46,test-47,test-48,test-49,test-50,test-51,test-52,test-53,test-54,test-55,----thread-org.springframework.kafka.KafkaListenerEndpointContainer#0-4-C-1started at time-2023/01/06 22:22:24 and time taken-132.69---"]

Время начала потоков различается с разницей около> 80 секунд между первым потоком и последним.

Любая идея, как решить эту проблему. Я хочу, чтобы поток работал почти одновременно (количество потоков может увеличиться до 15), что может улучшить прием огромных записей?

Кроме того, данные добавляются в разделы разного размера. Можно ли их разрешить?

Пользовательский скаляр GraphQL
Пользовательский скаляр GraphQL
Листовые узлы системы типов GraphQL называются скалярами. Достигнув скалярного типа, невозможно спуститься дальше по иерархии типов. Скалярный тип...
Как вычислять биты и понимать побитовые операторы в Java - объяснение с примерами
Как вычислять биты и понимать побитовые операторы в Java - объяснение с примерами
В компьютерном программировании биты играют важнейшую роль в представлении и манипулировании данными на двоичном уровне. Побитовые операции...
Поднятие тревоги для долго выполняющихся методов в Spring Boot
Поднятие тревоги для долго выполняющихся методов в Spring Boot
Приходилось ли вам сталкиваться с требованиями, в которых вас могли попросить поднять тревогу или выдать ошибку, когда метод Java занимает больше...
Полный курс Java для разработчиков веб-сайтов и приложений
Полный курс Java для разработчиков веб-сайтов и приложений
Получите сертификат Java Web и Application Developer, используя наш курс.
0
0
62
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Ваше started at time предположение неверно. На самом деле это не время запуска параллельных контейнеров, а время, когда они потребляют записи из назначенных им разделов. Таким образом, у вас может просто не быть данных в разделе для использования в данный момент. Более того, нет гарантии, что все разделы будут назначены потребителям одновременно. Таким образом, один потребитель может уже начать потреблять, а другой еще не получил закрепленные за ним разделы.

Чтобы равномерно распределить данные между разделами, вам нужно изучить абстракцию messageKey и Partitioner на стороне производителя.

Другие вопросы по теме