RabbitMQ @RabbitListener单队列未按指定路由键过滤消息问题

DevOps 小满

2026-8-26

问题根因

你对RabbitMQ的路由逻辑存在认知偏差：路由键匹配规则仅在交换机向队列投递消息时生效，消费者监听队列时不存在按路由键过滤消息的原生能力。
你代码中@QueueBinding注解的key参数，仅在Spring应用启动需要自动创建队列、建立队列和交换机的绑定关系时才会生效。你现在使用的是预先创建好的公共队列，该队列已经和交换机绑定了多个路由键，Spring启动时不会删除队列上已有的其他绑定关系，因此无论你在key参数中配置什么值，只要监听这个队列，就会拉取到队列内的所有消息。

重要提醒：你当前使用的单队列多团队共享架构本身不符合RabbitMQ最佳实践：RabbitMQ中队列内的消息会在多个消费者之间负载均衡，同一条消息只会投递给其中一个消费者，不会广播给所有消费者。如果你直接确认（ack）不属于你团队路由键的消息，会直接导致该消息丢失，对应负责的团队永远无法收到这条消息。
在无法修改现有架构、不能创建独立队列的约束下，只能在消费端实现路由键判断逻辑，对不属于自己的消息执行拒绝重回队列操作，让消息重新投递给其他消费者，直到被对应负责的团队消费。这种方案会产生额外的消息投递开销，若某类路由键对应的消费者全部下线，对应消息会在队列中反复弹跳，阻塞其他正常消息消费，属于架构约束下的折中方案。

解决方案

优先选择手动判断的实现方式，逻辑直观可控，不会因Spring版本差异出现兼容问题。

方案1：消费端手动判断路由键（推荐）

配置监听器为手动确认模式，在方法内获取消息的路由键做判断，非本团队的消息执行nack重回队列，本团队的消息正常处理后确认：

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "queue", durable = "true"),
        exchange = @Exchange(value = "exchange", autoDelete = "false", type = "topic"),
        key = "abc_rk"
), ackMode = "MANUAL")
public void consumeMessagesFromRabbitMQ(Message message, Channel channel, @Payload Request request) throws Exception {
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    String receivedRoutingKey = message.getMessageProperties().getReceivedRoutingKey();
    try {
        // 非本团队负责的路由键，拒绝消息并重回队列，供其他团队的消费者拉取
        if (!"abc_rk".equals(receivedRoutingKey)) {
            channel.basicNack(deliveryTag, false, true);
            return;
        }
        // 以下是本团队消息的业务处理逻辑
        System.out.println("Start:Request from RabbitMQ: " + request);
        Thread.sleep(10000L);
        System.out.println("End:Request from RabbitMQ: " + request);
        // 业务处理完成，手动确认消息
        channel.basicAck(deliveryTag, false);
    } catch (InterruptedException e) {
        // 业务异常根据实际需求决定是否重回队列，示例为不重回队列，若配置了死信交换机会进入死信队列
        channel.basicNack(deliveryTag, false, false);
        throw e;
    }
}

方案2：使用@RabbitListener条件过滤+自定义错误处理器

如果不想在业务方法内写判断逻辑，可以使用注解的condition属性做SpEL匹配，同时自定义错误处理器处理不匹配的消息，避免消息被直接丢弃：

自定义错误处理器，对不匹配条件的消息执行重投：

import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import com.rabbitmq.client.Channel;

public class RoutingKeyFilterErrorHandler implements RabbitListenerErrorHandler {
    @Override
    public Object handleError(Message message, Channel channel, org.springframework.messaging.Message<?> messagingMessage, ListenerExecutionFailedException e) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 条件不匹配导致的方法找不到异常，执行nack重投
        if (e.getMessage().contains("no matching listener method")) {
            channel.basicNack(deliveryTag, false, true);
            return null;
        }
        // 其他业务异常按实际需求处理，示例为不重投
        channel.basicNack(deliveryTag, false, false);
        throw new AmqpRejectAndDontRequeueException("Business process failed", e);
    }
}

将错误处理器注册为Spring Bean，在注解中引用并配置过滤条件：

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "queue", durable = "true"),
        exchange = @Exchange(value = "exchange", autoDelete = "false", type = "topic"),
        key = "abc_rk"
), condition = "headers['amqp_receivedRoutingKey'] == 'abc_rk'", errorHandler = "routingKeyFilterErrorHandler")
public void consumeMessagesFromRabbitMQ(Request request) throws InterruptedException {
    System.out.println("Start:Request from RabbitMQ: " + request);
    Thread.sleep(10000L);
    System.out.println("End:Request from RabbitMQ: " + request);
}

补充说明

现有代码中@QueueBinding配置的key = "abc_rk"在队列已存在的场景下不会修改原有绑定关系，没有实际作用，可以保留也可以删除，不影响消费逻辑。
消费端过滤重投的方案存在性能和可靠性隐患，后续如果架构允许调整，优先为每个团队创建独立队列绑定对应路由键，由RabbitMQ原生完成路由，既不会有消息丢失风险，也没有额外的投递开销。

内容的提问来源于stack exchange，提问作者Anonymous

RabbitMQ @RabbitListener单队列未按指定路由键过滤消息问题

方案1：消费端手动判断路由键（推荐） #

方案2：使用@RabbitListener条件过滤+自定义错误处理器 #

方案1：消费端手动判断路由键（推荐）

方案2：使用@RabbitListener条件过滤+自定义错误处理器