spring-boot - 将 Spring AMQP 消费者/生产者服务迁移到 Spring Stream 源时出现问题

Question

我正在迁移一个 Spring Boot 微服务，它使用服务器 A 上 3 个 RabbitMQ 队列中的数据，将其保存到 Redis 中，最后将消息生成到服务器 B 上不同 RabbitMQ 中的交换中，以便另一个微服务可以使用这些消息。此流程工作正常，但我想使用 RabbitMQ 绑定(bind)器将其迁移到 Spring Cloud Stream。所有 Spring AMQP 配置都是在属性文件中自定义的，没有使用 spring 属性来创建连接、队列、绑定(bind)等...

我的第一个想法是在 Spring Cloud Stream 中设置两个绑定(bind)，一个连接到服务器 A(消费者)，另一个连接到服务器 B(生产者)，并将现有代码迁移到处理器，但我放弃了它，因为它似乎如果使用多个 Binder ，则无法设置连接名称，并且我需要添加多个绑定(bind)以从服务器 A 的队列中使用，并且 BindingRoutingKey 属性不支持值列表(我知道它可以按照解释以编程方式完成 here )。

因此，我决定仅重构与生产者相关的代码部分，以通过 RabbitMQ 使用 Spring Cloud Stream，因此相同的微服务应通过 Spring AMQP 从服务器 A(原始代码)使用，并应通过 Spring Cloud Stream 生成到服务器 B .

我发现的第一个问题是 Spring Cloud Stream 中的 NonUniqueBeanDefinitionException，因为 org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory bean 使用 handlerMethodFactory 和 integrationMessageHandlerMethodFactory 名字。

org.springframework.beans.factory.NoUniqueBeanDefinitionException: No qualifying bean of type 'org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory' available: expected single matching bean but found 2: handlerMethodFactory,integrationMessageHandlerMethodFactory
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveNamedBean(DefaultListableBeanFactory.java:1144)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveBean(DefaultListableBeanFactory.java:411)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBean(DefaultListableBeanFactory.java:344)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.getBean(DefaultListableBeanFactory.java:337)
    at org.springframework.context.support.AbstractApplicationContext.getBean(AbstractApplicationContext.java:1123)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.injectAndPostProcessDependencies(StreamListenerAnnotationBeanPostProcessor.java:317)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(StreamListenerAnnotationBeanPostProcessor.java:113)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:862)
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:877)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:549)
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:743)
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:390)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:312)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1214)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1203)

似乎前一个 bean 是由 Spring AMQP 创建的，后者是由 Spring Cloud Stream 创建的，所以我创建了自己的主 bean:

@Bean
@Primary
public MessageHandlerMethodFactory messageHandlerMethodFactory() {
    return new DefaultMessageHandlerMethodFactory();
}

现在应用程序可以启动，但输出 channel 是由服务器 A 而不是服务器 B 中的 Spring Cloud Stream 创建的。看来 Spring Cloud Stream 配置正在使用 Spring AMQP 创建的连接，而不是使用其自己的配置。

Spring AMQP的配置是这样的:

@Bean
public SimpleRabbitListenerContainerFactory priceRabbitListenerContainerFactory(
 ConnectionFactory consumerConnectionFactory) {
  return
    getSimpleRabbitListenerContainerFactory(
      consumerConnectionFactory,
      rabbitProperties.getConsumer().getListeners().get(LISTENER_A));
}

@Bean
public SimpleRabbitListenerContainerFactory maxbetRabbitListenerContainerFactory(
  ConnectionFactory consumerConnectionFactory) {
    return
      getSimpleRabbitListenerContainerFactory(
        consumerConnectionFactory,
       rabbitProperties.getConsumer().getListeners().get(LISTENER_B));
}

@Bean
public ConnectionFactory consumerConnectionFactory() throws Exception {
  return
    new CachingConnectionFactory(
      getRabbitConnectionFactoryBean(
        rabbitProperties.getConsumer()
      ).getObject()
    );
}

private SimpleRabbitListenerContainerFactory getSimpleRabbitListenerContainerFactory(
  ConnectionFactory connectionFactory,
  RabbitProperties.ListenerProperties listenerProperties) {
    //return a SimpleRabbitListenerContainerFactory set up from external properties
}

/**
 * Create the AMQ Admin.
 */
@Bean
public AmqpAdmin consumerAmqpAdmin(ConnectionFactory consumerConnectionFactory) {
  return new RabbitAdmin(consumerConnectionFactory);
}

/**
 * Create the map of available queues and declare them in the admin.
 */
@Bean
public Map<String, Queue> queues(AmqpAdmin consumerAmqpAdmin) {
  return
    rabbitProperties.getConsumer().getListeners().entrySet().stream()
      .map(listenerEntry -> {
        Queue queue =
          QueueBuilder
            .nonDurable(listenerEntry.getValue().getQueueName())
            .autoDelete()
            .build();

          consumerAmqpAdmin.declareQueue(queue);

          return new AbstractMap.SimpleEntry<>(listenerEntry.getKey(), queue);
      }).collect(
        Collectors.toMap(
          AbstractMap.SimpleEntry::getKey,
          AbstractMap.SimpleEntry::getValue
        )
      );
}

/**
 * Create the map of available exchanges and declare them in the admin.
 */
@Bean
public Map<String, TopicExchange> exchanges(AmqpAdmin consumerAmqpAdmin) {
  return
    rabbitProperties.getConsumer().getListeners().entrySet().stream()
      .map(listenerEntry -> {
        TopicExchange exchange =
          new TopicExchange(listenerEntry.getValue().getExchangeName());

        consumerAmqpAdmin.declareExchange(exchange);

        return new AbstractMap.SimpleEntry<>(listenerEntry.getKey(), exchange);
      }).collect(
        Collectors.toMap(
          AbstractMap.SimpleEntry::getKey,
          AbstractMap.SimpleEntry::getValue
        )
      );
}

/**
 * Create the list of bindings and declare them in the admin.
 */
@Bean
public List<Binding> bindings(Map<String, Queue> queues, Map<String, TopicExchange> exchanges, AmqpAdmin consumerAmqpAdmin) {
  return
    rabbitProperties.getConsumer().getListeners().keySet().stream()
      .map(listenerName -> {
        Queue queue = queues.get(listenerName);
        TopicExchange exchange = exchanges.get(listenerName);

        return
          rabbitProperties.getConsumer().getListeners().get(listenerName).getKeys().stream()
            .map(bindingKey -> {
              Binding binding = BindingBuilder.bind(queue).to(exchange).with(bindingKey);

              consumerAmqpAdmin.declareBinding(binding);

              return binding;
            }).collect(Collectors.toList());
      }).flatMap(Collection::stream)
      .collect(Collectors.toList());
}

消息监听器是:

@RabbitListener(
  queues="${consumer.listeners.LISTENER_A.queue-name}",
  containerFactory = "priceRabbitListenerContainerFactory"
)
public void handleMessage(Message rawMessage, org.springframework.messaging.Message<ModelPayload> message) {
   // call a service to process the message payload
}

@RabbitListener(
  queues="${consumer.listeners.LISTENER_B.queue-name}",
  containerFactory = "maxbetRabbitListenerContainerFactory"
)
public void handleMessage(Message rawMessage, org.springframework.messaging.Message<ModelPayload> message) {
  // call a service to process the message payload
}

属性:

#
# Server A config (Spring AMQP)
#
consumer.host=server-a
consumer.username=
consumer.password=
consumer.port=5671
consumer.ssl.enabled=true
consumer.ssl.algorithm=TLSv1.2
consumer.ssl.validate-server-certificate=false
consumer.connection-name=local:microservice-1
consumer.thread-factory.thread-group-name=server-a-consumer
consumer.thread-factory.thread-name-prefix=server-a-consumer-
# LISTENER_A configuration
consumer.listeners.LISTENER_A.queue-name=local.listenerA
consumer.listeners.LISTENER_A.exchange-name=exchangeA
consumer.listeners.LISTENER_A.keys[0]=*.1.*.*
consumer.listeners.LISTENER_A.keys[1]=*.3.*.*
consumer.listeners.LISTENER_A.keys[2]=*.6.*.*
consumer.listeners.LISTENER_A.keys[3]=*.8.*.*
consumer.listeners.LISTENER_A.keys[4]=*.9.*.*
consumer.listeners.LISTENER_A.initial-concurrency=5
consumer.listeners.LISTENER_A.maximum-concurrency=20
consumer.listeners.LISTENER_A.thread-name-prefix=listenerA-consumer-
# LISTENER_B configuration
consumer.listeners.LISTENER_B.queue-name=local.listenerB
consumer.listeners.LISTENER_B.exchange-name=exchangeB
consumer.listeners.LISTENER_B.keys[0]=*.1.*
consumer.listeners.LISTENER_B.keys[1]=*.3.*
consumer.listeners.LISTENER_B.keys[2]=*.6.*
consumer.listeners.LISTENER_B.initial-concurrency=5
consumer.listeners.LISTENER_B.maximum-concurrency=20
consumer.listeners.LISTENER_B.thread-name-prefix=listenerB-consumer-

#
# Server B config (Spring Cloud Stream)
#
spring.rabbitmq.host=server-b
spring.rabbitmq.port=5672
spring.rabbitmq.username=
spring.rabbitmq.password=

spring.cloud.stream.bindings.outbound.destination=microservice-out
spring.cloud.stream.bindings.outbound.group=default

spring.cloud.stream.rabbit.binder.connection-name-prefix=local:microservice

所以我的问题是:是否可以在同一个 Spring Boot 应用程序代码中使用通过 Spring AMQP 使用来自 RabbitMQ 的数据并通过 Spring Cloud Stream RabbitMQ 将消息生成到不同的服务器？如果是的话，有人可以告诉我我做错了什么吗？

Spring AMQP版本是Boot版本2.1.7(2.1.8-RELEASE)提供的版本，Spring Cloud Stream版本是Spring Cloud train Greenwich.SR2(2.1.3.RELEASE)提供的版本。

编辑

我能够通过多个配置属性而不是默认属性来配置 Binder 。因此，通过此配置，它可以工作:

#
# Server B config (Spring Cloud Stream)
#
spring.cloud.stream.binders.transport-layer.type=rabbit
spring.cloud.stream.binders.transport-layer.environment.spring.rabbitmq.host=server-b
spring.cloud.stream.binders.transport-layer.environment.spring.rabbitmq.port=5672
spring.cloud.stream.binders.transport-layer.environment.spring.rabbitmq.username=
spring.cloud.stream.binders.transport-layer.environment.spring.rabbitmq.password=

spring.cloud.stream.bindings.stream-output.destination=microservice-out
spring.cloud.stream.bindings.stream-output.group=default

不幸的是，尚无法在多个绑定(bind)器配置中设置连接名称:A custom ConnectionNameStrategy is ignored if there is a custom binder configuration .

无论如何，我仍然不明白为什么在使用 Spring AMQP 和 Spring Cloud Stream RabbitMQ 时上下文似乎是“混合的”。为了使实现正常工作，仍然需要设置一个主 MessageHandlerMethodFactory bean。

编辑

我发现导致 NoUniqueBeanDefinitionException 的原因是微服务本身正在创建一个 ConditionalGenericConverter 供 Spring AMQP 部分用来反序列化来自服务器 A 的消息。

我删除了它并添加了一些MessageConverter。现在问题已经解决，不再需要@Primary bean。

Answer 1

不相关，但是

1. consumerAmqpAdmin.declareQueue(queue);

您永远不应该在 @Bean 定义中与代理进行通信；现在处于应用程序上下文生命周期还为时过早。它可能会起作用，但是 YMMV；此外，如果代理不可用，它也会阻止您的应用程序启动。

最好定义包含队列、 channel 、绑定(bind)列表的 Declarables 类型的 bean，并且管理员将在连接首次成功打开时自动声明它们。请参阅引用手册。

我从未见过 MessageHandlerFactory 问题； Spring AMQP 声明没有这样的 bean。如果您可以提供一个展示该行为的小型示例应用程序，那将会很有用。

我将看看能否找到解决连接名称问题的方法。

编辑

我找到了解决连接名称问题的方法；它需要一些反射(reflection)，但它确实有效。我建议你打开new feature request against the binder请求一种在使用多个绑定(bind)器时设置连接名称策略的机制。

无论如何；这是解决方法...

@SpringBootApplication
@EnableBinding(Processor.class)
public class So57725710Application {

    public static void main(String[] args) {
        SpringApplication.run(So57725710Application.class, args);
    }

    @Bean
    public Object connectionNameConfigurer(BinderFactory binderFactory) throws Exception {
        setConnectionName(binderFactory, "rabbit1", "myAppProducerSide");
        setConnectionName(binderFactory, "rabbit2", "myAppConsumerSide");
        return null;
    }

    private void setConnectionName(BinderFactory binderFactory, String binderName,
            String conName) throws Exception {

        binderFactory.getBinder(binderName, MessageChannel.class); // force creation
        @SuppressWarnings("unchecked")
        Map<String, Map.Entry<Binder<?, ?, ?>, ApplicationContext>> binders =
                (Map<String, Entry<Binder<?, ?, ?>, ApplicationContext>>) new DirectFieldAccessor(binderFactory)
                    .getPropertyValue("binderInstanceCache");
        binders.get(binderName)
                .getValue()
                .getBean(CachingConnectionFactory.class).setConnectionNameStrategy(queue -> conName);
    }

    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public String listen(String in) {
        System.out.println(in);
        return in.toUpperCase();
    }

}

和

spring.cloud.stream.binders.rabbit1.type=rabbit
spring.cloud.stream.binders.rabbit1.environment.spring.rabbitmq.host=localhost
spring.cloud.stream.binders.rabbit1.environment.spring.rabbitmq.port=5672
spring.cloud.stream.binders.rabbit1.environment.spring.rabbitmq.username=guest
spring.cloud.stream.binders.rabbit1.environment.spring.rabbitmq.password=guest

spring.cloud.stream.bindings.output.destination=outDest
spring.cloud.stream.bindings.output.producer.required-groups=outQueue
spring.cloud.stream.bindings.output.binder=rabbit1

spring.cloud.stream.binders.rabbit2.type=rabbit
spring.cloud.stream.binders.rabbit2.environment.spring.rabbitmq.host=localhost
spring.cloud.stream.binders.rabbit2.environment.spring.rabbitmq.port=5672
spring.cloud.stream.binders.rabbit2.environment.spring.rabbitmq.username=guest
spring.cloud.stream.binders.rabbit2.environment.spring.rabbitmq.password=guest

spring.cloud.stream.bindings.input.destination=inDest
spring.cloud.stream.bindings.input.group=default
spring.cloud.stream.bindings.input.binder=rabbit2

和