rabbitmq无法连接问题

这篇具有很好参考价值的文章主要介绍了rabbitmq无法连接问题。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

背景

        最近一个项目使用了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

到了这里,关于rabbitmq无法连接问题的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • [Spring boot] Spring boot 整合RabbitMQ实现通过RabbitMQ进行项目的连接

     🍳作者:天海奈奈 💭眼过千遍不如手锤一遍:推荐一款模拟面试,斩获大厂 o f f e r ,程序员的必备刷题平台 − − 牛客网  👉🏻点击开始刷题之旅 目录 什么是RabbitMQ   消息队列:接受并转发消息,类似于快递公司 消息队列的优点 消息队列的特性 RabbitMQ特点 RabbitMQ核

    2024年01月24日
    浏览(84)
  • RabbitMQ连接断开自动重连问题解决

    RabbitMQ是一个功能强大的消息代理,被广泛用于构建分布式应用程序。然而,有时候在使用RabbitMQ时会遇到连接断开的问题,这可能会导致消息传递中断和应用程序的不可用性。在本文中,我们将探讨如何解决RabbitMQ连接断开的问题,并提供相应的源代码示例。 当使用RabbitMQ时

    2024年02月05日
    浏览(55)
  • 登录RabbitMQ Management时提示不是私密连接问题

    情况:我是使用虚拟机docker安装的: 之后登录访问RabbitMQ管理页面后就弹出身份验证: 之前通过重装解决,这次又遇到且重装也没解决问题,排查后发现是安装启动mq时用户没创建成功。 进入docker容器查看用户 发现只有一个本guest用户,原来是自己的用户创建失败了,于是重

    2024年02月11日
    浏览(41)
  • windows环境下RabbitMQ无法访问http://localhost:15672的问题

    1.问题 成功安装RabbitMQ后,RabbitMQ管理模块的插件也启动了但是还是无法访问http://localhost:15672。 网上很多回答都是,如下: 其实在打开节点的时候就提示了问题: 在dos下使用rabbitmqctl status查看原因 查看原因发现是Erlang新版本的cookie位置换了 异常解释来源于:https://www.cnblogs.com/h

    2024年02月11日
    浏览(51)
  • STM32CubeIDE无法连接STLinkV2的一个可能原因

    前些日子买了个新的STLINKV2 , ( 老的被我家猫猫一泡尿送上西天了) 刚开始还挺正常的, 后来有一天就突然间发现无法调试了. 网上尝试了各种方法, 各种操作. 百度了好久. 最后好在我自己研究出来了. 我使用的是 STM32CubeIDE 这个开发工具 调试选项如下图 其它的不需要修改, 然后

    2024年02月05日
    浏览(42)
  • Docker:rabbitmq启动镜像后访问15672端口无法显示管理界面问题解决

    在Linux Ubuntu 18.04上通过Docker安装配置Rabbitmq后,按照网上提供的教程经验,发现打开 http://{ip}+15672无法打开相关的Rabbitmq Web管理平台界面 相关博客 Ubuntu安装Docker Docker安装Rabbitmq 这里无法通过15672端口获取相关文件可以划分为两个方向,笔者将其归类为外部因素和Rabbitmq内部原

    2024年02月11日
    浏览(45)
  • 浏览器无法连接网络问题

    电脑其他程序都能正常联网,但是所有的浏览器都无法联网,同时外部网站都能ping通 查看电脑Internet连接的问题报告显示: 该设备或资源(Web 代理)未设置为接受端口\\\"7890\\\"上的连接 。 经过检查发现不是IP地址设置和防火墙设置的问题,最终 采用重置网络设置解决了问题 。 重

    2024年02月13日
    浏览(43)
  • git无法clone项目连接失败(设置代理)

    代理服务器 就是你的电脑和互联网的中介。当您访问外网时(如http://google.com) , 你的请求首先转发到代理服务器,然后代理服务器替你访问外网,并将结果原封不动的给你的电脑,这样你的电脑就可以看到外网的内容。 路径如下: 你的电脑-代理服务器-外网 外网-代理服务器

    2024年02月13日
    浏览(48)
  • SSH连接问题解决方案 - 服务器无法连接

    SSH(Secure Shell)是一种常用的远程登录和执行命令的协议,它提供了安全的加密通信通道。然而,有时候在连接Linux服务器时可能会遇到无法连接的问题。本文将为您提供一些可能的解决方案。 确认服务器状态 首先,您需要确认服务器的状态。确保服务器正在运行,并且网络

    2024年02月03日
    浏览(47)
  • 解决git无法连接gitHub问题

    事情的起因 我在使用git push我的项目时出现下面问题 fatal: unable to access ‘https://github.com/xxx.git/’: Recv failure: Connection was reset 或者 fatal: unable to access ‘https://github.com/xxx.git/’: Failed to connect to github.com port 443 after 21090 ms: Couldn’t connect to server 翻译过来就是无法连接到github 分析

    2024年02月03日
    浏览(45)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包