持续的投票“只是”Kafka消费者的工作方式，Kafka协议下的工作方式。每个操作总是由客户机（拉模型）发起，而不是由代理（推模型）发起，在使用消息的情况下，代理将转换为轮询。使用javakafkaconsumerapi意味着拥有一个循环，使用一个调度器，或者使用java的任何技术来连续地执行代码，您必须处理它。其他框架，如spring或smallryeReact式消息传递，就是为您做的。他们将轮询循环隐藏到您的应用程序中，但最终总会有一个循环。。。Kafka就是这样工作的。

我试着使用一个调度程序和代码工作。

package com.kafka;
import java.text.ParseException;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.TimerTask;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
public class ScheduledTask extends TimerTask {
 Date now; // to display current time
    public void run( ) {
        now = new Date();
    System.out.println("Time is :" + now);
    String AlarmString=null;
    Properties props = new Properties();
     props.put("bootstrap.servers", "10.*.*.*:9092");
     props.put("group.id", "grp-1");
     props.put("enable.auto.commit", "true");
     props.put("auto.commit.interval.ms", "1000");
     props.put("session.timeout.ms", "30000");
     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
     consumer.subscribe(Arrays.asList("RAW_MH_RAN_SAM"));

    ConsumerRecords<String, String> records = consumer.poll(1000);
        //ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis(100));
         for (ConsumerRecord<String, String> record : records)
         {

             System.out.println("Consumer:=============== partition Id= " + record.partition() + "  offset = " + record.offset() + " value = " + record.value() + "=================");

             if (AlarmString==null && !(record.value().toString().contains("PR ALARM:")) ){

                 AlarmString=record.value(); 

                }
             else{
                    if( !(record.value().toString().contains("PR ALARM:"))   )

                    {
                        //System.out.println("record.value()   :::"+ record.value() );  
                        AlarmString=AlarmString+","+record.value();

                    }

                }

         }
     if (consumer != null) {
         System.out.println("Closing Connection");
         consumer.close();
      }

 }

}
//Simple consumer 
package com.kafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import com.google.gson.Gson;
import java.text.ParseException;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.Timer;
public class SimpleConsumer{
    public static void main(String[] args) 
{

    Timer time = new Timer();
    ScheduledTask st = new ScheduledTask(); // Instantiate SheduledTask class
    time.schedule(st, 0,60000); // Create Repetitively task for every 1 secs  
}

}