如何关闭/停止 Spring Cloud 流绑定的 RabbitMQ 队列答案

【问题标题】：How to Shutdown/Stop RabbitMQ queue of Spring Cloud stream bindings如何关闭/停止 Spring Cloud 流绑定的 RabbitMQ 队列
【发布时间】：2020-04-23 12:05:05
【问题描述】：

我想阻止rabbitmq 中的消费者在访问端点/prepare-for-shutdown 时从spring 云流绑定创建队列。请在下面找到配置，

在 pom.xml 中添加了依赖

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

Application.yml：

spring:
  cloud: 
    stream:
      bindings: 
        produceChannel: 
          binder: rabbit
          content-type: application/json
          destination: internal-exchange
        consumeChannel: 
          binder: rabbit
          content-type: application/json
          destination: internal-exchange
          group: small-queue
      rabbit:
        bindings:
          consumeChannel:
            consumer:
              autoBindDlq: true
              durableSubscription: true
              requeueRejected: false
              republishToDlq: true
              bindingRoutingKey: admin
          produceChannel: 
            producer:               
              routingKeyExpression: '"admin"'

示例.java

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;

public interface Sample{
    @Input("consumeChannel")
    SubscribableChannel consumeChannel();

    @Output("produceChannel")
    SubscribableChannel produceChannel();
}

使用 Spring Cloud 的 @StreamLinster 和 @EnableBinding 抽象实现了与 RabbitMQ 的集成，如下所示：

@EnableBinding(Sample.class)


@StreamListener("consumeChannel")
public void sampleMessage(String message) {
    // code
}

期待以编程方式停止 RabbitMQ 队列的消费者。提前致谢

【问题讨论】：

您可以通过执行器以编程方式执行此操作。见this answer。
但是在 BindingsEndpoint 类中调用gatherInputBindings() 时得到空值。我错过了任何代码/配置吗？
当我访问 /actuator/bindings 时，它返回获取空值。请推荐我@GaryRussell
我无法解释；它对我（和其他所有人）都很好；你一定有一些不正确的配置；如果您能提供一个展示此行为的小型、完整、可验证的示例，我可以看看。

标签： spring-boot rabbitmq spring-cloud-stream

【解决方案1】：

我分析了为什么通过调用执行器端点'/actuator/bindings'得到空值的问题

当点击执行器绑定端点时，它会调用 BindingsEndpoint.class 中的方法gatherInputBindings()。

在 BindingsEndpoint.java 中，从 inputBindingLifecycle 中获取绑定值

(Collection<Binding<?>>) new DirectFieldAccessor(inputBindingLifecycle).getPropertyValue("inputBindings");

在以下方法中，将空绑定列表设置为 inputBindings

在 InputBindingLifecycle.java 中，

void doStartWithBindable(Bindable bindable) {
    this.inputBindings = bindable.createAndBindInputs(bindingService);
}

在 Bindable.java 中，

default Collection<Binding<Object>> createAndBindInputs(BindingService adapter) {
        return Collections.<Binding<Object>>emptyList();
    }

请建议我解决这些问题是否需要更改任何依赖项或任何代码配置

【讨论】：