RabbitMQ断链重连的配置

懵懂

2024-01-07 帮助1人

- 问题描述：

自动删除的队列在网络断开恢复后不能自动创建并绑定关系。

- 解决办法：

在RabbitMQConfig中分别声明两个连接工厂类，其中一个用于RabbitAdmin声明的连接，另一个用于生产者和消费者的连接。其中用于RabbitAdmin声明的连接不设置连接、通道、恢复的事件监听，用于生产者和消费者的连接需要设置连接、通道、恢复的事件监听，并且在通道监听事件“onCreate(Channel channel, boolean b)”方法中调用声明队列和绑定关系。

- 参考示例：

CalcRabbitMQConfig

package com.cdjdgm.calc.config.rabbitmq;

import com.cdjdgm.calc.component.rabbitmq.CalcRabbitEventListener;

import com.cdjdgm.calc.component.rabbitmq.CalcRabbitMessageConverter;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.amqp.rabbit.annotation.EnableRabbit;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;

import org.springframework.amqp.rabbit.connection.ConnectionFactory;

import org.springframework.amqp.rabbit.core.RabbitAdmin;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Qualifier;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.context.annotation.Primary;

import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.concurrent.atomic.AtomicInteger;

/**

* RabbitMQ配置类

* @author ZhaoYong

@Configuration

@EnableRabbit

@ConditionalOnBean(name = "calcRabbitMQProperties")

public class CalcRabbitMQConfig {

private final Logger logger = LoggerFactory.getLogger(getClass());

private final String prefix = "calc";

private final AtomicInteger connectionNameStrategyCounter1 = new AtomicInteger();

private final AtomicInteger connectionNameStrategyCounter2 = new AtomicInteger();

@Primary

@Bean(name = "calcRabbitConnectionFactory")

public CachingConnectionFactory calcRabbitConnectionFactory(CalcRabbitMQProperties calcRabbitMQProperties,//

CalcRabbitEventListener calcRabbitEventListener) {

CachingConnectionFactory connectionFactory = new CachingConnectionFactory();

if (calcRabbitMQProperties.getHost().contains(":")) {

connectionFactory.setAddresses(calcRabbitMQProperties.getHost());

} else {

connectionFactory.setHost(calcRabbitMQProperties.getHost());

connectionFactory.setPort(calcRabbitMQProperties.getPort());

}

connectionFactory.setUsername(calcRabbitMQProperties.getUsername());

connectionFactory.setPassword(calcRabbitMQProperties.getPassword());

connectionFactory.setVirtualHost(calcRabbitMQProperties.getVirtualHost());

connectionFactory.setPublisherReturns(calcRabbitMQProperties.getPublisherReturns());

connectionFactory.setPublisherConfirmType(calcRabbitMQProperties.getPublisherConfirmType());

connectionFactory.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(true);

connectionFactory.getRabbitConnectionFactory().setTopologyRecoveryEnabled(true);

connectionFactory.getRabbitConnectionFactory().setNetworkRecoveryInterval(5000L);

connectionFactory.setConnectionThreadFactory(new CustomizableThreadFactory("calc-rabbitmq-"));

connectionFactory.addConnectionListener(calcRabbitEventListener);

connectionFactory.addChannelListener(calcRabbitEventListener);

connectionFactory.setRecoveryListener(calcRabbitEventListener);

connectionFactory.setConnectionNameStrategy(connectionFactory1 -> "[" prefix "]ConnectionFactory" //

":" connectionNameStrategyCounter1.getAndIncrement());

return connectionFactory;

}

@Bean(name = "calcRabbitConnectionFactoryAdmin")

public CachingConnectionFactory calcRabbitConnectionFactoryAdmin(CalcRabbitMQProperties calcRabbitMQProperties) {