带有 RabbitMQ 的 Spring Cloud Spring 服务连接器答案

【问题标题】：Spring Cloud Spring Service Connector with RabbitMQ带有 RabbitMQ 的 Spring Cloud Spring 服务连接器
【发布时间】：2018-05-12 05:56:22
【问题描述】：

我使用 Spring cloud Spring 服务连接器连接 CloudFoundry 上的 Rabbitmq 服务。

public class CloudConfig extends AbstractCloudConfig {

    @Bean
    public ConnectionFactory rabbitFactory()
    {
         return connectionFactory().rabbitConnectionFactory();
    }
}

但我需要声明一个 CachingConnectionFactory 并将其 PublisherConfirms 设置为 true。因为我们在向队列发送消息时需要使用 publisherConfirm 来检查 ack。我不知道如何注入从云弹簧服务连接器获得的 connectionFactory。或者我们如何处理这种情况。

【问题讨论】：

标签： spring-amqp spring-rabbit spring-cloud-connectors

【解决方案1】：

您可以重新配置连接器创建的 CCF，如下所示：

@Bean
public SmartInitializingSingleton factoryConfigurer() {
    return new SmartInitializingSingleton() {

        @Autowired
        private CachingConnectionFactory connectionFactory;

        @Override
        public void afterSingletonsInstantiated() {
            this.connectionFactory.setPublisherConfirms(true);
        }
    };
}

您必须确保在应用程序上下文完全初始化之前不要执行任何 RabbitMQ 操作（无论如何这是最佳实践）。

【讨论】：

@RabbitListener 会影响设置吗？
你的意思是我可以这样配置？而且我还需要配置@Bean public ConnectionFactory rabbitFactory() { return connectionFactory().rabbitConnectionFactory(); }
Scott 的解决方案更好。
今天，我在我的项目中尝试了sn-p代码。但我发现它直到现在才起作用。但是确认真假。

【解决方案2】：

documentation 包含自定义连接器提供的连接详细信息的示例。

在你的情况下，你应该能够做这样的事情：

@Bean
public RabbitConnectionFactory rabbitFactory() {
    Map<String, Object> properties = new HashMap<String, Object>();
    properties.put("publisherConfirms", true);

    RabbitConnectionFactoryConfig rabbitConfig = new RabbitConnectionFactoryConfig(properties);
    return connectionFactory().rabbitConnectionFactory(rabbitConfig);
}

【讨论】：

酷 - 我不知道。注意有一个typo in the documentation。实际返回类型为org.springframework.amqp.rabbit.connection.ConnectionFactory。
感谢 Gary 的问题，我会看看。
非常感谢您的帮助。
@ScottFrederick 我还有一个问题想问你。正如你所说，我实施了我们的项目。但它抛出了一个属性 publisherConfirms 无法写入的异常。没有 setter。我使用 1.6.9v amqp 库。你能帮我一把吗？
@Maxwell 异常是否显示它试图设置 publisherConfirms 的类？我假设它将设置在org.springframework.amqp.rabbit.connection.CachingConnectionFactory 上，它确实有一个设置器：docs.spring.io/spring-amqp/api/org/springframework/amqp/rabbit/…。

【解决方案3】：

这是兔子模板

@Bean
public RabbitTemplate rabbitTemplate() {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setMandatory(true);
    template.setMessageConverter(new Jackson2JsonMessageConverter());
    template.setConfirmCallback((correlationData, ack, cause) -> {
        if (!ack) {
            System.out.println("send message failed: " + cause + correlationData.toString());
        } else {
            System.out.println("Publisher Confirm" + correlationData.toString());
        }
    });
    return template;
}

这是 spring-cloud 配置：

@Bean
public ConnectionFactory rabbitConnectionFactory() {
    Map<String, Object> properties = new HashMap<String, Object>();
    properties.put("publisherConfirms", true);
    RabbitConnectionFactoryConfig rabbitConfig = new RabbitConnectionFactoryConfig(properties);
    return connectionFactory().rabbitConnectionFactory(rabbitConfig);
}

当我使用此发件人发送消息时。结果不是预期的。

@Component
public class TestSender {

@Autowired
private RabbitTemplate rabbitTemplate;

@Scheduled(cron = "0/5 * *  * * ? ")
public void send() {
System.out.println("===============================================================");
    this.rabbitTemplate.convertAndSend(EXCHANGE, "routingkey", "hello world",
            (Message m) -> {
                m.getMessageProperties().setHeader("tenant", "aaaaa");
                return m;
            }, new CorrelationData(UUID.randomUUID().toString()));
    Date date = new Date();
    System.out.println("Sender Msg Successfully - " + date);
}

}

【讨论】：