如何有选择地从AMQP(rabbitmq)队列中删除消息?

您目前无法在rabbitmq(或者更一般地说,在amqp)中自动执行此操作。但是,这里有一个简单的解决方法。

假设您要发送三种类型的消息:xs、ys和zs。如果我正确理解了您的问题,那么当X消息到达时,您希望代理忽略所有其他尚未传递的X消息。

这在rabbitmq中很容易做到:

• 生产者声明了三个队列:X、Y和Z(它们自动绑定到默认交换,其名称作为路由键,这正是我们想要的)。
• 发布消息时,生产者首先清除相关队列(因此,如果它发布X消息,则首先清除X队列);这有效地删除了过时的消息,
• 消费者只需从它想要的队列中消费(x表示x消息,y表示y消息等);从它的角度来看,它只需要做一个基本的操作。获取下一个相关的消息。

这意味着当两个生产者大约同时发送同一类型的消息时存在竞争条件。结果是,一个队列可能同时有两个(或更多)消息,但是由于消息的数量由生产者的数量上限,并且由于多余的消息在下一次发布时被清除,所以这不应该是一个很大的问题。

总而言之,此解决方案与最佳解决方案只有一个额外步骤,即在发布X类型的消息之前清除队列X。

如果您需要任何帮助来设置此配置,那么最好的咨询位置是rabbitmq讨论邮件列表。

不需要消息队列,需要键值数据库。例如,您可以使用redis或tokyo tyrant获得一个简单的网络可访问的键值数据库。或者使用memcache。

每种消息类型都是一个键。当使用相同的键编写新消息时,它将覆盖以前的值,因此此数据库的读卡器将永远无法获取过期信息。

此时,您只需要一个消息队列来确定读取密钥的顺序(如果这很重要的话)。否则,只需继续扫描数据库。如果继续扫描数据库,最好将数据库放在读卡器附近,以减少网络流量。

我可能会做这样的事 key: typecode value: lastUpdated, important data

然后我会发送包含 typecode, lastUpdated 这样,读卡器可以将该密钥的lastupdated与上次从数据库中读取的密钥进行比较,并跳过读取,因为它们已经是最新的了。

如果您真的需要使用AMQP来实现这一点,那么可以使用rabbitmq和自定义交换类型,特别是最后一个值缓存交换。这里是示例代码 https://github.com/squaremo/rabbitmq-lvc-plugin

这个问题因其标题而具有很高的知名度。通过描述可以找到更具体的场景。 因此,对于希望从队列中实际删除下一条(记住FIFO)消息的用户,可以使用rabbitmqadmin并发出以下命令:

rabbitmqadmin get queue=queuename requeue=false count=1

这个命令基本上是在消耗消息,什么也不做。带有备份消息标志的完整命令可能如下所示。确保根据您的要求添加任何其他参数。

sudo python rabbitmqadmin -V virtualhostname -u user -p pass get queue=queuename requeue=false count=1 payload_file=~/origmsg

如果您只想从队列中删除前n条消息,那么它似乎也可以从rabbitmq Web UI中工作。

• 从“队列”选项卡中选择队列,向下滚动到“获取消息”部分。
• 设置参数“requeue=no”和要从队列中删除的消息数
• 按“获取消息”按钮