如何手动调用spring kafkaListener方法

Question

我创建了一个 java 类和方法有 @KafkaListener。但我希望只有当我从 UI 中单击按钮时才需要调用此方法。当我从 UI 中单击一个按钮时，调用会转到控制器类，而我想从控制器类调用这个类。怎么能做到这一点？ 我没有添加任何类级别的注释。

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") && 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();
    }

Answer 1

现在我想搜索所有这些分区以查找特定员工的数据是否存在。

那么@KafkaListener 不适合你。当请求完成时，看起来您的要求是按需的。 @KafkaListener 本身就是一个东西：它依赖于下面的不断轮询循环。它完全不依赖于最终用户的请求。

您需要考虑根据您在 WEB 请求中收到的数据即时创建KafkaConsumer。请参阅ConsumerFactory，而不是直接使用它来创建这些消费者。然后您可以在该消费者上调用poll() 并最终关闭它。这已经超出了 Apache Kafka 的 Spring 范围：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