rabbitmq - 使用 spring-amqp 和 rabbitmq 实现带退避的非阻塞重试

Question

我正在寻找一种使用 spring amqp 和 Rabbit MQ 使用退避策略实现重试的好方法，但要求是不应阻塞监听器(因此可以自由处理其他消息)。我在这里看到了一个类似的问题，但它不包括“退出”的解决方案:
RabbitMQ & Spring amqp retry without blocking consumers
我的问题是:

重试时默认的 spring-retry 实现会阻塞线程吗？ implementation in github表示确实如此。

如果上述假设成立，那么实现此目的的唯一方法是实现一个单独的重试队列(DLQ？)，并为每条消息设置一个 TTL(假设我们不想在退避间隔内阻塞线程)。

如果我们采用上述方法(DLQ 或单独的队列)，每次重试尝试不需要单独的队列吗？如果我们只使用 1 个队列进行重试，同一个队列将包含 TTL 从最小重试间隔到最大重试间隔的消息，如果队列前面的消息具有最大 TTL，则它后面的消息不会即使它有最小 TTL 也被捡起。这是根据 Rabbit MQ TTL 文档 here (见注意事项):

是否有另一种方法来实现非阻塞退避重试机制？

添加一些配置信息以帮助解决@garyrusel:
队列配置:

    <rabbit:queue name="regular_requests_queue"/>
    <rabbit:queue name="retry_requests_queue">
        <rabbit:queue-arguments>
            <entry key="x-dead-letter-exchange" value="regular_exchange" />
        </rabbit:queue-arguments>
    </rabbit:queue>

    <rabbit:direct-exchange name="regular_exchange">
        <rabbit:bindings>
            <rabbit:binding queue="regular_requests_queue" key="regular-request-key"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <rabbit:direct-exchange name="retry_exchange">
        <rabbit:bindings>
            <rabbit:binding queue="retry_requests_queue"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <bean id="retryRecoverer" class="com.testretry.RetryRecoverer">
         <constructor-arg ref="retryTemplate"/>
         <constructor-arg value="retry_exchange"/>
    </bean>

    <rabbit:template id="templateWithOneRetry" connection-factory="connectionFactory" exchange="regular_exchange" retry-template="retryTemplate"/>
    <rabbit:template id="retryTemplate" connection-factory="connectionFactory" exchange="retry_exchange"/>

    <bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate">
        <property name="retryPolicy">
            <bean class="org.springframework.retry.policy.SimpleRetryPolicy">
                <property name="maxAttempts" value="1"/>
            </bean>
        </property>
    </bean>

Answer 1

是

通 4 ...

您可以将重试最大尝试次数 = 1 与 RepublishMessageRecoverer 的子类一起使用并实现 additionalHeaders添加，说重试计数标题。

然后，您可以为每次尝试重新发布到不同的队列。

恢复器的结构并没有真正发布到不同的队列(我们应该改变它)，所以您可能需要编写自己的恢复器并委托(delegate)给几个 RepublishMessageRecoverer 之一.

考虑 contributing您对框架的解决方案。