我在Java中使用官方的RabbitMQ amqp-client 来处理消息，并把它们发送到一个外部REST API。当前，我把 autoAck 禁用，并通过调用 basicNack，传入 requeue = true 来处理失败。

问题： 当外部API不可用时，requeue = true 会导致消息被立即重新投递。这会形成一个无尽循环，消耗大量CPU，并且因为重试之间没有延迟，日志会被连接错误淹没。

// Simplified logic
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    long deliveryTag = delivery.getEnvelope().getDeliveryTag();
    try {
        // ... JSON parsing and API call ...
        if (response.statusCode() == 200) {
            channel.basicAck(deliveryTag, false);
        } else {
            // This causes immediate redelivery loop
            channel.basicNack(deliveryTag, false, true);
        }
    } catch (Exception e) {
        channel.basicNack(deliveryTag, false, true);
    }
};

我考虑过的方案：

• 使用 Thread.sleep()：我想避免这种做法，因为它会阻塞通道/连接，影响其他消息。
• Spring RabbitMQ：我正在寻找一个使用 原生Java客户端 的解决方案，而不需要将整个项目迁移到Spring。

问题： 如何使用 amqp-client 库实现带有延迟的重试机制（如指数退避），或使用TTL和死信交换机（DLX）将消息移动到一个“等待”队列？如果有代码示例或处理外部API停机的模式，将不胜感激。

解决方案

我尝试在本地构建该项目，以便看看你的问题，这个思路浮现了：

1. 设置 requeue = false，让消息不会立即进入work.queue。
2. 创建单独的队列，用TTL配置的时间来存放消息。
3. 根据外部API调用的结果，将消息路由到你选择的队列。
4. 你也可以添加一个 重试计数器，通过将消息移动到TTL更长的队列来增加重试间隔；一旦达到重试上限，就将消息移到一个专门用于死信的队列。

简而言之，我们得到的是：
work.queue → (fail) → retry.5s.queue → (TTL expired) → work.queue

换句话说，我们不会通过 basicNack(..., true) 立即重新处理消息，而是将其暂时放入一个单独的队列。TTL到期后，RabbitMQ会通过死信交换机（DLX）自动把消息返回到主队列。

简单来说，消费者的逻辑可能如下：

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    long deliveryTag = delivery.getEnvelope().getDeliveryTag();

    try {
        // ... JSON parsing and external API call ...

        if (response.statusCode() >= 200 && response.statusCode() < 300) {
            channel.basicAck(deliveryTag, false);
            return;
        }

        if (isRetriable(response.statusCode())) {
            int retryCount = getRetryCount(delivery);

            if (retryCount >= 3) {
                // retry limit reached -> move to dead queue
                channel.basicPublish(
                        "dead.exchange",
                        "dead",
                        delivery.getProperties(),
                        delivery.getBody()
                );
            } else {
                // publish to retry queue instead of requeue=true
                channel.basicPublish(
                        "retry.exchange",
                        resolveRetryRoutingKey(retryCount),
                        delivery.getProperties(),
                        delivery.getBody()
                );
            }

            // acknowledge original message after republishing
            channel.basicAck(deliveryTag, false);
        } else {
            // non-retriable error -> move directly to dead queue
            channel.basicPublish(
                    "dead.exchange",
                    "dead",
                    delivery.getProperties(),
                    delivery.getBody()
            );

            channel.basicAck(deliveryTag, false);
        }

    } catch (Exception e) {
        int retryCount = getRetryCount(delivery);

        if (retryCount >= 3) {
            channel.basicPublish(
                    "dead.exchange",
                    "dead",
                    delivery.getProperties(),
                    delivery.getBody()
            );
        } else {
            channel.basicPublish(
                    "retry.exchange",
                    resolveRetryRoutingKey(retryCount),
                    delivery.getProperties(),
                    delivery.getBody()
            );
        }

        channel.basicAck(deliveryTag, false);
    }
};

补充方法：

private static boolean isRetriable(int statusCode) {
    return statusCode == 429 || statusCode >= 500;
}

private static String resolveRetryRoutingKey(int retryCount) {
    return switch (retryCount) {
        case 0 -> "retry.5s";
        case 1 -> "retry.30s";
        default -> "retry.5m";
    };
}

与其从你自定义的头部读取重试次数，不如使用 x-death 头部，当消息从重试队列返回时，RabbitMQ会自动添加该头部：

private static int getRetryCount(Delivery delivery) {
    Map<String, Object> headers = delivery.getProperties().getHeaders();

    if (headers == null) {
        return 0;
    }

    Object xDeathObj = headers.get("x-death");

    if (!(xDeathObj instanceof List<?> xDeathList) || xDeathList.isEmpty()) {
        return 0;
    }

    int totalRetries = 0;

    for (Object entryObj : xDeathList) {
        if (!(entryObj instanceof Map<?, ?> entry)) {
            continue;
        }

        Object queue = entry.get("queue");
        Object count = entry.get("count");

        if (queue instanceof String queueName
                && queueName.startsWith("retry.")
                && count instanceof Number number) {
            totalRetries += number.intValue();
        }
    }

    return totalRetries;
}

拓扑结构本身可能像这样：

channel.exchangeDeclare("work.exchange", "direct", true);
channel.exchangeDeclare("retry.exchange", "direct", true);
channel.exchangeDeclare("dead.exchange", "direct", true);

channel.queueDeclare("work.queue", true, false, false, null);
channel.queueDeclare("dead.queue", true, false, false, null);

Map<String, Object> retry5sArgs = new HashMap<>();
retry5sArgs.put("x-message-ttl", 5000);
retry5sArgs.put("x-dead-letter-exchange", "work.exchange");
retry5sArgs.put("x-dead-letter-routing-key", "work");

Map<String, Object> retry30sArgs = new HashMap<>();
retry30sArgs.put("x-message-ttl", 30000);
retry30sArgs.put("x-dead-letter-exchange", "work.exchange");
retry30sArgs.put("x-dead-letter-routing-key", "work");

Map<String, Object> retry5mArgs = new HashMap<>();
retry5mArgs.put("x-message-ttl", 300000);
retry5mArgs.put("x-dead-letter-exchange", "work.exchange");
retry5mArgs.put("x-dead-letter-routing-key", "work");

channel.queueDeclare("retry.5s.queue", true, false, false, retry5sArgs);
channel.queueDeclare("retry.30s.queue", true, false, false, retry30sArgs);
channel.queueDeclare("retry.5m.queue", true, false, false, retry5mArgs);

channel.queueBind("work.queue", "work.exchange", "work");
channel.queueBind("retry.5s.queue", "retry.exchange", "retry.5s");
channel.queueBind("retry.30s.queue", "retry.exchange", "retry.30s");
channel.queueBind("retry.5m.queue", "retry.exchange", "retry.5m");
channel.queueBind("dead.queue", "dead.exchange", "dead");

结果是如下的设置：

• work.queue 仅用于正常处理
• retry.*.queue 作为holding队列使用
• TTL到期后，RabbitMQ会自动把消息返回到work.queue
• 一旦达到重试上限，消息会被移动到dead.queue