如何从Spring Cloud Stream Kafka活页夹中的偏移量获得消息？

Question

Setting offset for partition MY_TOPIC-0 to the committed offset FetchPosition{offset=1076, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=broker.test.com:6667 (id: 1003 rack: /default-rack), epoch=2}}

public interface EventConsumer {

    @Input("my-group-id")
    SubscribableChannel consumeMessage();

}

侦听器类

@Slf4j
@Component
@RequiredArgsConstructor
@EnableBinding(EventConsumer.class)
public class EventListener {

     @StreamListener(target = "my-group-id")
     public void processMessage(Object msg) {
         log.info("*** MESSAGE: ***", msg);
         **do something**
         **save messages**
     }
}

在读取日志时，它甚至不会进入我为它放置记录器的侦听器类。对此有什么想法吗？

Answer 1

您没有指定从哪个主题消费

@KafkaListener(topics = "MY_TOPIC", groupId = "foo")
public void listenToYourTopic(String message) {
    System.out.println("Received Message in group foo from topic: " + message);
}

当需要为给定主题指定分区时：

@KafkaListener(
  topicPartitions = @TopicPartition(topic = "topicName",
  partitionOffsets = {
    @PartitionOffset(partition = "0", initialOffset = "0"), 
    @PartitionOffset(partition = "3", initialOffset = "0")}))
public void listenToPartition(
  @Payload String message, 
  @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
      System.out.println(
        "Received Message: " + message"
        + "from partition: " + partition);
}

很好的参考：https://www.baeldung.com/spring-kafka