У нас есть приложение Spring, которое имеет динамический прослушиватель очереди, подключающийся к очереди в rabbitmq.Допустим, у меня есть 5 слушателей, подключенных к 5 очередям из моего весеннего приложения в rabbitmq.
Теперь, если происходит сбой / сбой в работе сети, каждый раз первая из 5 подключенных очередей останавливается, пытаясь вернуться к rabbitmq.
Я отлаживал код через класс spring-amqp и обнаружил, что насоздавая соединение с rabbitmq (когда происходит сбой сети), он не может подключиться к нему и генерирует особое исключение org.springframework.amqp.AmqpIOException, которое не обрабатывается в функции повтора, так что эта очередь удаляется из списка очередей с повторными попытками.
Мой Основной класс:
@Slf4j
@SpringBootApplication(exclude = {ClientAutoConfiguration.class})
@EnableTransactionManagement
@EnableJpaRepositories(basePackages = "com.x.x.repositories")
@EntityScan(basePackages = "com.x.x.entities")
public class Main
{
@PostConstruct
void configuration()
{
TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
}
/**
* The main method.
*
* @param args the arguments
*/
public static void main(String[] args)
{
ConfigurableApplicationContext context = SpringApplication.run(Main.class, args);
RabbitMQListenerUtil queueRegisterUtil = context.getBean(RabbitMQListenerUtil.class);
try
{
queueRegisterUtil.registerSpecifiedListenerForAllInstance();
}
catch (Exception e)
{
log.error(e.getMessage(), e);
}
}
}
Класс, который используется для создания 5 потребителей / слушателей
/**
* The Class RabbitMQListenerUtil.
*/
@Component
@Slf4j
public class RabbitMQListenerUtil
{
@Autowired
private ApplicationContext applicationContext;
public void registerSpecifiedListenerForAllInstance()
{
try
{
log.debug("New Listener has been register for instane name : ");
Thread.sleep(5000);
registerNewListener("temp1");
registerNewListener("temp2");
registerNewListener("temp3");
registerNewListener("temp4");
registerNewListener("temp5");
}
catch (Exception e)
{
}
}
/**
* This method will add new listener bean for given queue name at runtime
*
* @param queueName - Queue name
* @return Configurable application context
*/
public void registerNewListener(String queueName)
{
AnnotationConfigApplicationContext childAnnotaionConfigContext = new AnnotationConfigApplicationContext();
childAnnotaionConfigContext.setParent(applicationContext);
ConfigurableEnvironment environmentConfig = childAnnotaionConfigContext.getEnvironment();
Properties listenerProperties = new Properties();
listenerProperties.setProperty("queue.name", queueName + "_queue");
PropertiesPropertySource pps = new PropertiesPropertySource("props", listenerProperties);
environmentConfig.getPropertySources().addLast(pps);
childAnnotaionConfigContext.register(RabbitMQListenerConfig.class);
childAnnotaionConfigContext.refresh();
}
}
Класс, который создает динамический слушатель для потребителя очереди
/**
* The Class RabbitMQListenerConfig.
*/
@Configuration
@Slf4j
@EnableRabbit
public class RabbitMQListenerConfig
{
/** The Constant ALLOW_MESSAGE_REQUEUE. */
private static final boolean ALLOW_MESSAGE_REQUEUE = true;
/** The Constant MULTIPLE_MESSAGE_FALSE. */
private static final boolean MULTIPLE_MESSAGE_FALSE = false;
/**
* Listen.
*
* @param msg the msg
* @param channel the channel
* @param queue the queue
* @param deliveryTag the delivery tag
* @throws IOException Signals that an I/O exception has occurred.
*/
@RabbitListener(queues = "${queue.name}")
public void listen(Message msg, Channel channel, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException
{
int msgExecutionStatus = 0;
try
{
String message = new String(msg.getBody(), StandardCharsets.UTF_8);
log.info(message);
}
catch (Exception e)
{
log.error(e.toString());
log.error(e.getMessage(), e);
}
finally
{
ackMessage(channel, deliveryTag, msgExecutionStatus);
}
}
/**
* Ack message.
*
* @param channel the channel
* @param deliveryTag the delivery tag
* @param msgExecutionStatus the msg execution status
* @throws IOException Signals that an I/O exception has occurred.
*/
protected void ackMessage(Channel channel, long deliveryTag, int msgExecutionStatus) throws IOException
{
if (msgExecutionStatus == Constants.MESSAGE_DELETE_FOUND_EXCEPTION)
{
channel.basicNack(deliveryTag, MULTIPLE_MESSAGE_FALSE, ALLOW_MESSAGE_REQUEUE);
}
else
{
channel.basicAck(deliveryTag, MULTIPLE_MESSAGE_FALSE);
}
}
/**
* Bean will create from this with given name.
*
* @param name - Queue name-
* @return the queue
*/
@Bean
public Queue queue(@Value("${queue.name}") String name)
{
return new Queue(name);
}
/**
* RabbitAdmin Instance will be created which is required to create new Queue.
*
* @param cf - Connection factory
* @return the rabbit admin
*/
@Bean
public RabbitAdmin admin(ConnectionFactory cf)
{
return new RabbitAdmin(cf);
}
}
Журнал приложения:
https://pastebin.com/NQWdmdTH
Я проверял это несколько раз, и каждый раз, когда моя первая подключенная очередь не могла подключиться.
========================= ОБНОВЛЕНИЕ 1 =============================
Код для повторного подключения остановленного потребителя: https://pastebin.com/VnUrhdLP