RabbitMQ

Question

我有以下 PHP 应用程序。将用户注册发布到消息队列。 Java 应用程序从该队列中读取并导入它。希望下图能够描述它。我只处理 Java 方面的事情。 json 条消息已存在于队列中。

路线（Java消费方）。

@Component
public class SignUpRouting {

  errorHandler(deadLetterChannel("rabbitmq://signUpDeadLetter.exchange?username=etc..").useOriginalMessage());

  from("rabbitmq://phpSignUp.exchange?username=etc....")
            .routeId("signUpRoute")
            .processRef("signUpProcessor")
            .end();
  //....

处理器..

@Component
public class SignupProcessor implements Processor {

    private ObjectMapper mapper = new ObjectMapper();

    @Override
    public void process(Exchange exchange) throws Exception {

        String json = exchange.getIn().getBody(String.class);
        SignUpDto dto = mapper.readValue(json, SignUpDto.class);

        SignUp signUp = new SignUp();
        signUp.setWhatever(dto.getWhatever());
        //etc....

        // save record
        signUpDao.save(signUp);
    }
}

My question is this.. What should I do I do when the Processor fails to import the message.

假设有一个 DAO 异常。数据字段可能太长或导入的格式不正确。我不想丢失消息。我想查看错误并重试导入。但我不想每 30 秒重试一次消息。

我想我需要创建另一个队列..一个死信队列并且每 6 小时无限期地重试消息？..然后我会查看日志看到错误并上传修复和消息将被重新处理？

我将如何实施？还是我走错路了？

EDIT I have tried setting deadLetterExchange to see if would get things on the right direction... However it errors and says queue cannot be non null

 rabbitmq://phpSignUp.exchange?username=etc...&deadLetterExchange=signUpDeadLetter.exchange

Answer 1

您可以使用 onException 来捕获异常，如果有异常，消息将路由到死信交换，这里是 Spring DSL 中的示例：

<onException useOriginalMessage="true">
            <exception>java.sql.SQLException</exception>
            <redeliveryPolicy asyncDelayedRedelivery="true" maximumRedeliveries="1" redeliveryDelay="1000"/>

            <inOnly uri="rabbitmq://localhost/dead.msgs?exchangeType=fanout&amp;
                    autoDelete=false&amp;
                    bridgeEndpoint=true"/>
</onException>

Answer 2

下面是一个使用死信头的例子：

        <from uri="rabbitmq://localhost/youexchange?queue=yourq1&amp;
            exchangeType=topic&amp;
            routingKey=user.reg.*&amp;
            deadLetterExchange=dead.msgs&amp;
            deadLetterExchangeType=topic&amp;
            deadLetterRoutingKey=dead.letters&amp;
            deadLetterQueue=dead.letters&amp;
            autoAck=false&amp;
            autoDelete=false"/>

          <!--We can use onException to make camel to retry, and after that, dead letter queue are the fallback-->
        <onException useOriginalMessage="true">
            <exception>java.lang.Exception</exception>
            <redeliveryPolicy asyncDelayedRedelivery="true" maximumRedeliveries="3" redeliveryDelay="5000"/>
        </onException>

我们需要关闭autoAck并设置deadLetterQueue，那么如果有Exception抛出，消息会是死信队列。使用onException，我们可以控制camel将消息丢到死信队列之前的重试。

RabbitMQ - Apache Camel Reading Messages 如何处理失败的消息

RabbitMQ - Apache Camel Reading Messages what to do with failed messages

java

dead-letter