背景
最近一个项目使用了rabbitmq作为消息队列,进行异步解耦操作,因涉及到数据的一致性问题,设置了手动应答和持久化功能。开发过程中一切顺利,然而天将降大任于斯人也必先苦其心智老其筋骨,饿其体肤,空乏其身,好吧偏题了。。。。在最终的测试运行中发现一些偶尔会有消息无法发送的情况,有时候1、2周出现,有时候1、2小时出现完全没有规律。本文记载了相关问题并继续处理。
Rabbit配置
1、设置publisher-confirm-type和publisher-returns发布确认属性,其中publisher-confirm-type有三类值:NONE、CORRELATED、SIMPLE
- NONE:禁用发布确认模式,是默认值;
- CORRELATED:发布消息成功到交换器后会触发回调方法;
- SIMPLE:触发回调方法,并在发布消息成功后,调用waitForConfirms或waitForConfirmsOrDie方法等待返回发送结果。
2、配置acknowledge-mode为manual手动确认消息
- acknowledge-mode 三种值
- none 自动确认,收到消息就通知broker,是默认值
- manual 手动确认
- auto 根据异常情况确认
rabbitmq:
host: ******
port: 5672
publisher-confirm-type: correlated
publisher-returns: true
listener:
simple:
acknowledge-mode: manual
retry:
enabled: true
RabbitTemplate配置
public RabbitSend(RabbitTemplate rabbitTemplate) {
super();
this.rabbitTemplate = rabbitTemplate;
this.rabbitTemplate.setMandatory(true);
this.rabbitTemplate.setReturnsCallback(data -> {
try {
Thread.sleep(SLEEP_TIME);
logger.info("消息发送重试=====>{}", data);
} catch (Exception e) {
logger.error("发送失败", e);
}
});
this.rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
logger.info("消息发送确认成功====>{}", correlationData);
} else {
logger.info("消息发送失败=====>{}", correlationData);
}
});
}
发送消息
发送消息这里设置消息的持久化属性。
public void routeSend(String message, String exchange, String routingKey) {
Message msg = this.setMessage(message);
logger.info("开始发送消息");
rabbitTemplate.convertAndSend(exchange, routingKey, msg, new CorrelationData());
logger.info("消息发送完成");
}
private Message setMessage(String json) {
MessageProperties messageProperties = new MessageProperties();
messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);//持久化
return new Message(json.getBytes(), messageProperties);
}
消费消息
消费消息这里采用手动应答的方式,同时如果出现异常将消息移到队尾。
try {
//处理消息,并手动应答
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
logger.error("消费失败:" + e.getMessage());
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
channel.basicPublish(message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(),
MessageProperties.PERSISTENT_TEXT_PLAIN, JSON.toJSONBytes(message));
}
问题
悲剧的时刻来了,系统运行过程中会出现莫名奇妙的发送消息失败,并且程序假死。于是只能去调试源码,通过日志的打印发现“消息发送完成”这行日志没有打印出来,因此基本确定是rabbitTemplate.convertAndSend这行有问题,跟踪进去:
//1128行
public void convertAndSend(String exchange, String routingKey, final Object object,
@Nullable CorrelationData correlationData) throws AmqpException {
send(exchange, routingKey, convertMessageIfNecessary(object), correlationData);
}
//1063行execute()
public void send(final String exchange, final String routingKey,
final Message message, @Nullable final CorrelationData correlationData)
throws AmqpException {
execute(channel -> {
doSend(channel, exchange, routingKey, message,
(RabbitTemplate.this.returnsCallback != null
|| (correlationData != null && StringUtils.hasText(correlationData.getId())))
&& isMandatoryFor(message),
correlationData);
return null;
}, obtainTargetConnectionFactory(this.sendConnectionFactorySelectorExpression, message));
}
//2136行 doExecute(action, connectionFactory)
@Nullable
private <T> T execute(final ChannelCallback<T> action, final ConnectionFactory connectionFactory) {
if (this.retryTemplate != null) {
try {
return this.retryTemplate.execute(
(RetryCallback<T, Exception>) context -> doExecute(action, connectionFactory),
(RecoveryCallback<T>) this.recoveryCallback);
}
catch (RuntimeException e) { // NOSONAR catch and rethrow needed to avoid next catch
throw e;
}
catch (Exception e) {
throw RabbitExceptionTranslator.convertRabbitAccessException(e);
}
}
else {
return doExecute(action, connectionFactory);
}
}
//ConnectionFactoryUtils.createConnection
private <T> T doExecute(ChannelCallback<T> action, ConnectionFactory connectionFactory) {
...
connection = ConnectionFactoryUtils.createConnection(connectionFactory,this.usePublisherConnection);
...
}
继续ConnectionFactoryUtils.java
public static Connection createConnection(final ConnectionFactory connectionFactory,
final boolean publisherConnectionIfPossible) {
if (publisherConnectionIfPossible) {
ConnectionFactory publisherFactory = connectionFactory.getPublisherConnectionFactory();
if (publisherFactory != null) {
return publisherFactory.createConnection();
}
}
return connectionFactory.createConnection();
}
跳转到CachingConnectionFactory.java
public final Connection createConnection() throws AmqpException {
if (this.stopped) {
throw new AmqpApplicationContextClosedException(
"The ApplicationContext is closed and the ConnectionFactory can no longer create connections.");
}
synchronized (this.connectionMonitor) {//罪魁祸首,最后阻塞在这里
if (this.cacheMode == CacheMode.CHANNEL) {
if (this.connection.target == null) {
this.connection.target = super.createBareConnection();
// invoke the listener *after* this.connection is assigned
if (!this.checkoutPermits.containsKey(this.connection)) {
this.checkoutPermits.put(this.connection, new Semaphore(this.channelCacheSize));
}
this.connection.closeNotified.set(false);
getConnectionListener().onCreate(this.connection);
}
return this.connection;
}
else if (this.cacheMode == CacheMode.CONNECTION) {
return connectionFromCache();
}
}
return null; // NOSONAR - never reach here - exceptions
}
经过漫长的调试最终发现阻塞在CachingConnectionFactory的721行synchronized (this.connectionMonitor)。connectionMonitor是一个Object对象,加了synchronized锁,但是在对connectionMonitor加锁的地方都打上断点后发现并没有哪里锁住了对象。。。
本文记录了rabbitmq的问题,期待大神能够提点。文章来源:https://www.toymoban.com/news/detail-475617.html
文章来源地址https://www.toymoban.com/news/detail-475617.html
到了这里,关于rabbitmq无法连接问题的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!