0

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

这是配置：

@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 中。如果我直接从其中发布到 DLQ，它确实如此handleMessage。
  2. convertSendAndReceive没有收到任何东西（可能是例外？）并等到超时发生，在我的情况下是 5 分钟。它可能是有意的，但对于 RPC 风格的调用来说，它相当奇怪。

我是否错过或配置错误？

4

1 回答 1

1

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

您拒绝在 上的传递，intakeQueue并且消息将被路由到其DLX/DLQ（如果已配置）。

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

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

您可以将 rabbit 模板配置为使用固定的回复队列，但是您必须提供一个回复侦听器容器，如文档中所述。

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

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

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

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

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

您还可以在 AMQP上使用 Spring Remoting，它将负责传播异常。

于 2015-11-22T15:19:59.513 回答