Kafka生产者与消费者问题

Question

1. 向Kafka推送100,000条消息
2. 在使用者使用所有100,000条消息之前，使用Ctrl-C关闭zookeeper和kafka服务（这是通过在consumer方法中使用thread.sleep(1000)来模拟的）。

发现

在zookeeper和kafka服务被关闭后，消费者继续在控制台上写消息。

问题

我如何使消费者从上次消费的消息的索引+1继续。

1. 向Kafka推送100,000条消息
2. 在使用者使用所有100,000条消息之前，使用Ctrl-C关闭zookeeper和kafka服务（这是通过在consumer方法中使用thread.sleep(1000)来模拟的）。
3. 终止使用消息的spring boot应用程序
4. 提供动物园管理员和Kafka服务。
5. 启动负责使用消息的Spring Boot应用程序。

发现

使用者从一开始就使用所有的消息，忽略上次使用的消息。

期望值

@Configuration
@EnableKafka
public class KafkaConsumerConfig
{
    @Bean
    public Map<String, Object> consumerConfigs()
    {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "basic-group");
        return props;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory()
    {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
            kafkaListenerContainerFactory()
    {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(100);
        return factory;
    }
}

KafKaProducerConfig

@Configuration
    public class KafkaProducerConfig
    {
        @Bean
        public ProducerFactory<Integer, String> producerFactory()
        {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }

        @Bean
        public Map<String, Object> producerConfigs()
        {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            return props;
        }

        @Bean
        public KafkaTemplate<Integer, String> kafkaTemplate()
        {
            return new KafkaTemplate<>(producerFactory());
        }
    }

Kafka生产者

@Component
public class KafkaProducer
{

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void sendMessage(String message, final String topicName)
    {

        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName, message);

        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>()
        {

            @Override
            public void onSuccess(SendResult<String, String> result)
            {
                System.out.println("Sent message=[" + message
                        + "] with offset=[" + result.getRecordMetadata().offset() + "]");
            }

            @Override
            public void onFailure(Throwable ex)
            {
                System.out.println("Unable to send message=["
                        + message + "] due to : " + ex.getMessage());
            }
        });
    }
}

Kafka孔苏梅尔

@Component
public class KafkaConsumer
{

    @KafkaListener(id = "basic", topics = "test-1", clientIdPrefix = "test-prefix-id", autoStartup = "true", concurrency = "3")
    public void multipleTransactionNotification(@Payload final String message)
    {
        try
        {
            Thread.sleep(1000);
            System.out.println(message);
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }
}

测试应用程序

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

   @Bean
    public CommandLineRunner commandLineRunner(ApplicationContext ctx, KafkaProducer producer, NIPKafkaConsumer consumer) {
        return args -> {

            for (int i = 0; i < 100_000; i++)
            {
                producer.sendMessage("New Message-" + i, "test-1");
            }

        };
    }    

}

Answer 1

Spring Kafka执行kafkaconsumer.poll()来获取记录，然后为每个记录调用侦听器。

因此，如果批处理包含100条记录，则使用实际的KafKalistener进行整个批处理将需要100秒。

使用者提交偏移量以知道从何处恢复。

默认行为是每5秒自动提交一次偏移量。我的猜测是你在消费者做出补偿之前就杀死了它，这就是为什么它从一开始就重新开始的原因。

请参阅enable.auto.commit和auto.commit.interval.msKafka使用者配置。

也可以手动提交偏移量。