RabbitMq:替换重复的消息


RabbitMq: Replace duplicated messages

我正在使用RabbitMq为注册的网络钩子提交数据。

基地信息:如果在系统中创建了联系人,则消息将放入队列中,使用者稍后会将挂钩数据发送到注册的 url。

对于我的问题:联系人有可能在 5 秒内更新两次,并且两条消息仍在队列中。但我希望,如果第二条消息排队,第一条消息将被删除。

我知道我无法手动删除邮件。但是,是否可以以某种方式在消息上设置一个id,如果两个具有相同id的消息在同一队列中,则会自动删除/替换第一个消息?仅向 URL 发送一个请求。我知道您可以在消息本身上设置消息 ID。但是我没有找到任何东西来取代旧的。

我的PHP代码(简体):

    $connection = new AMQPConnection('localhost', 5672, 'test', 'test');
    $channel = $connection->channel();
    $channel->queue_declare(self::QUEUE_NAME, false, true, false, false);
    $data = array(
        'model' => get_class($subject),
        'id' => $subject->getId(),
        'event' => $event->getName()
    );
    $messageProperties = array(
        'message_id' => get_class($subject) . '-' . $subject->getId()
    );
    $channel->basic_publish(new AMQPMessage(json_encode($data), $messageProperties), '', self::QUEUE_NAME);
    $channel->close();
    $connection->close();

顺便说一句,我正在使用php amqplib https://github.com/videlalvaro/php-amqplib。

感谢您的帮助弗洛

RabbitMQ 不会以这种方式删除/过滤消息。您必须在应用程序级别执行此操作,可能使用诸如布隆过滤器之类的东西。

您可以使用唯一的邮件 ID 标记每封邮件。使用应用程序应保留已处理的入站消息 ID 的优化列表(在线程安全HashMapJava) 或Dictionary.NET) 实现中)。

如果消息到达,并且已经处理过(消息 ID 存在于存储的已处理消息 ID 列表中),它将被忽略(或者应发出礼貌的"请稍候"样式响应),从而保留幂等性。