Блог

Я создал класс Java, а метод имеет @KafkaListener. Но я хочу, чтобы этот метод вызывался только тогда, когда я нажимаю кнопку из пользовательского интерфейса. Когда я нажимаю кнопку из пользовательского интерфейса, вызов переходит к классу контроллера, а из класса контроллера я хочу вызвать этот класс. Как это можно сделать? Я не добавлял никаких аннотаций на уровне класса.

 public class MessageListener implements ConsumerSeekAware {

   
    private String topicName;
     
     private int partition;
     
     private long beginOffset;
     
     private long endOffset;
 
  
    public MessageListener(String topicName, int partition, long beginOffset, long endOffset) {
        super();
        this.topicName = topicName;
        this.partition = partition;
        this.beginOffset = beginOffset;
        this.endOffset = endOffset;
    }
 

    @KafkaListener(topics = "topicName", 
    concurrency = "20",
    clientIdPrefix = "clientId-Test",
    groupId = "clientId-group")
    public void handleMessage(final ConsumerRecord<Object, Object> consumerRecord) {
        System.out.println(consumerRecord);
    }
 

}

Вариант 2:

        public List<String> searchMessages(String topicName, int partitionNo, long beginOffset, long endOffset) {

    List<String> filteredMessages = new ArrayList<>();
    TopicPartition tp = new TopicPartition("topicName", partitionNo);
    Properties clusterOneProps = kafkaConsumerConfig.getConsumerProperties();
    KafkaConsumer<String, Object> consumer = new KafkaConsumer<>(clusterOneProps);

    try {
        consumer.subscribe(Collections.singletonList("topicName"), new ConsumerRebalanceListener() {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            // TODO Auto-generated method stub
        }
        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            // TODO Auto-generated method stub
            
            consumer.seek(tp, beginOffset); 
        }
    }); 
    Thread.sleep(100);    
    boolean flag = true;
    System.out.println("search started......from offset is " beginOffset);
    while(flag) {   
    ConsumerRecords<String, Object> crs = consumer.poll(Duration.ofMillis(100L));
     for (ConsumerRecord<String, Object> record : crs) {
              // search criteria
               if(record.value().toString().contains("01111") amp;amp; record.value().toString().contains("2021-11-06")) {
                   System.out.println("founddddddddddddddddddddddddddddddddddddddd " record.offset());
                   filteredMessages.add(record.value().toString());
               }
               if (record.offset() == endOffset) {
                   flag = false;
                   break;
               }
        }
     }
     System.out.println("doneeeeeeeeeeeeeeeee");
    }catch(Exception e) {
        e.printStackTrace();
    }finally {
        consumer.close();
    }
 

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

Тогда @KafkaListener это не для вас. Похоже, что ваше требование выполняется по требованию, когда запрос выполнен. Это @KafkaListener вещь сама по себе: она полагается на постоянный цикл опроса внизу. Это полностью не зависит от запросов конечного пользователя.

Вам нужно подумать о создании KafkaConsumer «на лету» на основе данных, полученных в веб-запросе. ConsumerFactory Вместо этого смотрите, как использовать его напрямую для создания этих потребителей. Затем вы можете вызвать poll() этого потребителя и закрыть его в конце. Однако это уже выходит за рамки Spring для Apache Kafka: https://docs.spring.io/spring-kafka/docs/current/reference/html/#connecting

Или… Вы можете обновить до последней spring-kafka-2.8.0 версии и использовать KafkaTemplate.receive() API: https://docs.spring.io/spring-kafka/docs/current/reference/html/#kafka-template-receive