基于 Spring Boot 的应用程序中的死信路由

Question

我已经阅读了有关该主题的多个答案，似乎我拥有的配置应该可以正常工作，但由于某种原因它不能正常工作。

这里是配置：

@Bean Queue intakeQueue(String name) { return new Queue(name, true); }

@Bean Exchange dlx(String name) { return new DirectExchange(name); }

@Bean Queue dlq(String name) { return new Queue(name, false, false, true); }

@Bean
Binding dlb(Exchange dlx, Queue dlq, Queue reply) {
    return BindingBuilder.bind(dlq).to(dlx).with(reply.getName()).noargs();
}

@Bean
Queue replyQueue(String name, Exchange dlx) {
    Map<String, Object> args = new HashMap<>();
    args.put("x-dead-letter-exchange", dlx.getName());
    args.put("x-dead-letter-routing-key", name);
    return new Queue(name, true, false, false, args);
}

RabbitMQ UI 显示回复队列具有DLX 和DLK 属性。

我发送类似的消息

this.rabbit.convertSendAndReceive(intakeQueue, obj, message -> {
    message.getMessageProperties().setPriority(10);
    return message;
});

消息处理程序在收到消息后立即抛出AmqpRejectAndDontRequeueException。这样做是故意的，仅用于测试。我从重试建议开始，但由于它没有产生任何结果，所以我简化了测试用例。

public Object handleMessage(Object obj) throws IOException {
    throw new AmqpRejectAndDontRequeueException("Testing retries!");
}

我现在看到两个问题：

1. 在抛出 ARADRE 后，消息永远不会出现在 DLQ 中。不过，如果我直接从handleMessage 发布到 DLQ 就可以了。
2. convertSendAndReceive 没有收到任何回复（可能是例外？）并等待超时发生，在我的情况下是 5 分钟。这可能是有意的，但对于 RPC 风格的调用来说，这很奇怪。

我是否遗漏或配置错误？

Answer 1

从这个意义上说，它不是 RPC；在监听器中抛出异常不会被传播回发送者。

您拒绝在intakeQueue 上发送邮件，邮件将被路由到其 DLX/DLQ（如果已配置）。

对于您的测试用例，您应该查看那个 DLQ（如果有的话），而不是回复队列的 DLQ。

RabbitMQ 不知道队列关系。您不显示容器或 RabbitTemplate 配置，但默认情况下使用直接回复路由（使用“特殊”内部队列）。

您可以将rabbit模板配置为使用固定的回复队列，但是您必须提供回复侦听器容器as described in the documentation。

如果发送者在收到回复时超时，则模板会抛出一个ARADRE，因此在这种情况下回复将是死信。

为了测试这一点，在你的监听器中睡眠超过超时时间，然后回复；然后您应该会看到回复转到回复队列的 DLQ。

如果要将异常传播给调用者，则需要将其作为handleMessage 返回值返回。

目前没有重新抛出此类异常的逻辑，但是，您必须在代码中执行此操作（检测回复是异常并重新抛出它）。

当然，异常类型必须是Serializable。

您也可以使用Spring Remoting over AMQP，它将负责传播异常。