标签: rabbitmq

刚刚升级到RabbitMQ的新版本 - 2.3.1 - 现在出现以下错误:

PRECONDITION_FAILED unknown delivery tag 1  

...随后关闭频道.这适用于较旧的RabbitMQ,没有客户端更改.

在应用程序行为方面:

当App A想要向App b发送异步消息并从B接收答案时,这是算法:

1. 应用A生成唯一ID并将其放入消息对象中
2. 然后,App A订阅一个新的队列,队列名称和路由键都等于uuid.
3. 应用B打开消息,进行一些计算并使用收到的路径将结果返回到通道.
4. 应用A获得答案并关闭队列.

到目前为止,1.7.0一切都很顺利.2.3.1出了什么问题？

当应用程序A调用时basicPublish(),应用程序B立即抛出以下异常:

com.rabbitmq.client.ShutdownSignalException: channel error; reason: {#method<channel.close>(reply-code=406,reply-text=PRECONDITION_FAILED - unknown delivery tag 1,class-id=60,method-id=80),null,""}
    at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:191)
    at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:159)
    at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:110)
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:438)
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; reason: {#method<channel.close>(reply-code=406,reply-text=PRECONDITION_FAILED - unknown delivery tag 1,class-id=60,method-id=80),null,""}

我有一个场景,我需要执行一系列过程,每个步骤都在独立的应用程序中完成和缩放.我正在为所有交易所使用主题交换.当前拓扑结构如下:

P - > X - > Q - > C/P - > X - > Q - > C.

我们正在"版本化"我们的队列以处理影响消息结构的可能的需求变化.绑定可能看起来像这样:

step1.exchange绑定到带有绑定键step1.v1的step1.v1.queue

step1.exchange绑定到带有绑定键step1.v2的step1.v2.queue

还有其他与版本无关的绑定模式也使得主题交换成为合适的选择.但是,我们只能使用一个交换来完成同样的事情.

TLDR:当您的用例可以以任何方式工作时,使用多个主题交换而不是一个主题交换是否有益处？

如何使用两个不同的芹菜项目,它消耗来自单个RabbitMQ安装的消息.

通常,如果我使用不同的rabbitmq,这些脚本可以正常工作.但是在生产机器上,我需要为它们共享相同的RabbitMQ后端.

注意:由于一些约束,我无法合并现有的新项目,因此它将是两个不同的项目.

有没有办法通过简单的Curl获取rabbitmq中队列的大小(剩余消息)？

就像是 curl -xget http://host:1234/api/queue/test/stats

谢谢

我是RabbitMQ的新手,所以请原谅我的琐碎问题:

1)如果在RabbitMQ中进行集群,如果节点发生故障,则加载转移到另一个节点(不停止其他节点).同样,我们也可以在不停止集群中的现有节点的情况下向现有集群添加新的新节点.那是对的吗？

2)假设我们从一个rabbitMQ节点开始,并在其上创建100个队列.现在,制作人开始以更快的速度发送消息 要处理此负载,我们会添加更多节点并创建一个集群.但队列仅存在于第一个节点上.现在节点之间的负载如何平衡？如果我们需要添加更多队列,我们​​应该在哪个节点添加它们？或者我们可以使用负载均衡器添加它们.

提前致谢

我正在使用带有EasyNetQ库的C#中的RabbitMQ.我在这里使用了pub/sub模式.我仍然有一些问题希望有人可以帮助我:

1. 当消费消息时出现错误时,它会自动移动到错误队列中.如何实现重试(以便将其放回原始队列,当它无法处理X次时,它会移动到死信队列)？
2. 据我所知,总有1个错误队列用于转储来自所有其他队列的消息.如何为每种类型设置1个错误队列,以便每个队列都有自己的关联错误队列？
3. 如何轻松重试错误队列中的消息？我试过Hosepipe,但它只是将消息重新发布到错误队列而不是原始队列.我不太喜欢这个选项,因为我不想在控制台中摆弄.我最好只针对错误队列进行编程.

任何人？

我正在尝试为连接到RabbitMQ代理的Scala/Java应用程序创建集成测试.为了达到这个目的,我想要一个能够在每次测试之前开始和停止的AMQP的嵌入式代理.最初我尝试将ActiveMQ作为嵌入式代理引入AMQP,但是应用程序使用RabbitMQ只能说AMQP版本0.9.3而ActiveMQ需要AMQP版本1.0.

我可以使用另一个嵌入式代理来代替ActiveMQ吗？

我已经在我的应用程序中实现了rabbitmq并且它在Windows Server 2008服务器上运行,问题是erl.exe占用高CPU用量,有时它达到40-45%的CPU使用率,即使在理想情况下(当不处理任何队列时) CPU占用率至少需要4-15%.

什么可能是高CPU使用率的原因？我需要做任何设置或任何其他事情.

我有一个用例,需要按照发布/订阅消息传递模式在服务器和客户端之间进行实时通信.生产者将是java,节点等服务器,客户端将是 - java桌面应用程序,移动应用程序(android/ios),浏览器(javascript).

我已经探讨了下面讨论的许多选项,但我无法提出强大的可扩展解决方案.

使用案例:服务器将发布各种主题的通知/消息,订阅一组主题的所有客户端(java/js/ios)将实时获取这些消息.

我按照3种方法解决了这个问题1> socketIo/socketcluster 2>使用mosquitto/rabbitmq作为代理探索了mqtt协议.3>探索卡夫卡

主要目标是使该架构具有高度可扩展性,不仅可以同时拥有超过百万个客户端连接,而且每秒可以发布和消耗超过百万条消息.

第一种方法很简单,但它可以工作,但webSocket不是可扩展的解决方案.

第二种方法可行,但是rabbitmq将创建大量队列(百万个客户端的百万个队列),因为它为每个连接到它的客户端维护队列,而且RabbitMq没有高消息发布和消耗率,加上假设我们有一个集群RabbitMQ的节点那么只有一个节点用于处理请求和其他的用于高可用性,但不平行的消耗.

第三,我探索了kafka,它以我的基准着称,我使用kafka的高级java api在java中创建客户端,可以用来订阅kafka主题,并且发布到该主题的任何消息都可以实时传递给客户端.

所以我的问题是使用kafka客户端进行实时推送通知有多好,其中所有的Java桌面应用程序(可能是一百万)将包含这个kafka java客户端sdk并将订阅某些主题,在这里我对待每个客户端作为消费者群体.

而且,这里一个主要问题是,这卡夫卡的客户是规模大,由于其阶依赖,因此使用此客户端Android将不会是一个很好的选择,也是我不认为它会奏效.

MQTT擅长在这里,因为它有针对Android,JAVA,IOS等官方phao客户

此外,我还没有在网上使用kafka看到用于百万消费者的发布/订阅消息传递的示例,大多数人正在将其用于数据管道,例如:实时日志处理,向HDFS提供数据,分析引擎等,流处理.

主要问题是我如何使用mqtt协议(适用于android/ios/web/iot)和kafka作为消息代理(具有较高的发布/订阅率)并为此问题提出可扩展的解决方案.

我的用例在某种程度上也类似于uber,其中有数百万的android/ios设备(客户端),我们实际上可以看到我们在地图上所有汽车的实时移动,有没有人知道后面的架构是什么这些实时跟踪汽车.

最近,我在生产者/消费者队列系统上做了快速实现.

<?php
namespace Queue;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;    

class Amqp
{
    private $connection;
    private $queueName;
    private $delayedQueueName;
    private $channel;
    private $callback;

    public function __construct($host, $port, $login, $password, $queueName)
    {
        $this->connection = new AMQPStreamConnection($host, $port, $login, $password);
        $this->queueName = $queueName;
        $this->delayedQueueName = null;
        $this->channel = $this->connection->channel();
        // First, we need to make sure that RabbitMQ will never lose our queue.
        // In order to do so, we need to declare it as durable. To do so we pass
        // …