Spring Cloud Stream手动投票器Kafka

Question

我不想使用@KafKalistener或@StreamListener，但我想手动轮询Kafka。我正在使用spring-cloud-starter-stream-kafka库，我有以下Kafka制作人

  @Autowired
  private KafkaTemplate<byte[], byte[]> template;

  public void sendMessages() {
    IntStream.range(2)
             .forEach(val -> {
               template.send("kafka-topic", "hello".getBytes());
             });
  }

 @Autowired
  private ConsumerFactory consumerFactory;

  public void processKafkaRecords() throws InterruptedException {
    Consumer<byte[], byte[]> consumer = consumerFactory.createConsumer("0", "consumer-1");
    consumer.subscribe(Arrays.asList("kafka-topic"));
    ConsumerRecords<byte[], byte[]> poll = consumer.poll(Duration.ofMillis(1000));
    poll.forEach(record -> {
      log.info("record {}", record);
    });
  }

spring.cloud.stream.bindings.pollableInput.destination=kafka-topic
spring.cloud.stream.bindings.pollableInput.group=kafka-topic
spring.cloud.stream.bindings.pollableInput.consumer.batch-mode=true
spring.cloud.stream.bindings.pollableInput.consumer.header-mode=none
spring.cloud.stream.bindings.pollableInput.consumer.use-native-decoding=true

spring.cloud.stream.kafka.bindings.pollableInput.consumer.autoCommitOffset=false

Answer 1

可能有几个原因：

1. duration.ofmilis(1000)-尝试增加时间，除非您的客户端和kafka都运行在同一台计算机上，否则在某些情况下，1秒可能太低。因为poll(Duration)的文档说明如果超时到期，将返回一个空记录集
2. 如果您首先启动了生产者，然后启动了使用者，并且您没有将偏移量重置策略设置为“最早”，那么您将看不到任何消息，因为默认情况下，使用者将从最新的偏移量进行使用。因此，请尝试设置以下auto.offset.reset=aristy
3. 来自同一使用者组的另一个使用者可能正在运行，并且只有1个分区，或者使用者组已处于最后一个偏移量。在这种情况下，您可以尝试更改使用者组ID。