停止Spring Cloud Stream@StreamListener侦听，直到收到某个Spring事件

Question

我正在开发Camunda BPM Spring Boot应用程序。该应用程序使用Spring Cloud Stream从Rabbitmq队列中读取消息。一旦收到消息，应用程序就会调用Camunda中的进程实例。

如果在应用程序启动期间rabbitmq队列中已经存在消息，则云流侦听器甚至在初始化Camunda之前就开始读取消息。

是否可以停止云流侦听器侦听队列，直到触发某个事件为止？在本例中为PostDeployEvent。

我已经创建了一个示例应用程序供参考https://github.com/kpkurian/spring-cloud-stream-camunda

谢谢

Answer 1

Spring cloud stream（Kafka活页夹）添加了暂停和恢复消费者的方法

@SpringBootApplication
@EnableBinding(Sink.class)
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @StreamListener(Sink.INPUT)
    public void in(String in, @Header(KafkaHeaders.CONSUMER) Consumer<?, ?> consumer) {
        System.out.println(in);
        consumer.pause(Collections.singleton(new TopicPartition("myTopic", 0)));
    }

    @Bean
    public ApplicationListener<ListenerContainerIdleEvent> idleListener() {
        return event -> {
            System.out.println(event);
            if (event.getConsumer().paused().size() > 0) {
                event.getConsumer().resume(event.getConsumer().paused());
            }
        };
    }
}

请检查文档https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_usage_examples

但是我认为暂停方法有一些问题：https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/479

PS/您可以在示例侦听器中获取partion id和主题名称：

  @StreamListener(Sink.INPUT)
  public void in(String in,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
        @Header(KafkaHeaders.CONSUMER) Consumer<?, ?> consumer) {
    System.out.println(in);
    TopicPartition p = new TopicPartition(topic, partition);
    consumer.pause(Collections.singleton(p));
  }

或在errorChannel全局侦听器中

   @StreamListener("errorChannel")
  public void errorGlobal(Message<?> message) {
    Message<?> failedMessage = ((ErrorMessage)message).getOriginalMessage();
    Consumer consumer = (Consumer)failedMessage.getHeaders().get(KafkaHeaders.CONSUMER);
    int partition = (int) failedMessage.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID);
    String topic = (String) failedMessage.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC);
    TopicPartition p = new TopicPartition(topic, partition);
    // ?
    consumer.pause(Collections.singleton(p));
}

Answer 2

根据@OlegZhurakousky的建议

问题

RuntimeService是自动装配的，在启动应用程序时，假设所有服务、bean等都已完全初始化。如果它仍在经历初始化和启动过程，那么从Spring习惯用法的角度来看，它没有正确实现。

解决方案

使用自定义生命周期实现包装RuntimeService，直到执行其start（）方法确保RuntmeService准备就绪后才会返回。

我已经在示例github应用程序中实现了这一点