java - 防止AMQP队列消费者中的无限循环

Question

我想在 AMQP 队列监听器中实现 Web Flux 客户端。我试过这个:

@Component
public class TransactionGenesisAuthorizeListener {

    public TransactionResponseFactory transactionGenesisAuthorizeProcess(AuthorizeRequestFactory tf) throws Exception {

        AuthorizeResponse response = null;              

        try {           
                RestClient client = RestClientBuilder.builder()
                        .gatewayUrl(URL)
                        .build();

                Mono<AuthorizeResponse> result = client.executeAndReceiveAuthorize(request);
                response = result.block();

        return parseRawSuccessResponse(response);
    }

    private TransactionResponseFactory parseRawSuccessResponse(AuthorizeResponse response) {
        ................

        return obj;
    }

}

//网络客户端

public Mono<AuthorizeResponse> executeAndReceiveAuthorize(AuthorizeRequest transaction) {
        Mono<AuthorizeRequest> transactionMono = Mono.just(transaction);
        return client.post().uri(checkTrailingSlash(gatewayUrl) + token)
                .header(HttpHeaders.USER_AGENT, "Mozilla/5.0")
                .accept(MediaType.APPLICATION_XML)
                .contentType(MediaType.APPLICATION_XML)
                .body(transactionMono, AuthorizeRequest.class)
                .retrieve()
                .bodyToMono(AuthorizeResponse.class);
    }

错误日志

2019-08-27 19:42:09,280 INFO  [stdout] (processingTransactionGenesisAuthorizeContainer-1) 19:42:09.280 [processingTransactionGenesisAuthorizeContainer-1] WARN  o.s.a.r.l.SimpleMessageListenerContainer - Consumer raised exception, processing can restart if the connection factory supports it. Exception summary: org.springframework.amqp.rabbit.support.ConsumerCancelledException
2019-08-27 19:42:09,282 INFO  [stdout] (processingTransactionGenesisAuthorizeContainer-1) 19:42:09.282 [processingTransactionGenesisAuthorizeContainer-1] INFO  o.s.a.r.l.SimpleMessageListenerContainer - Restarting Consumer@6e537459: tags=[[]], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,10), conn: Proxy@70156185 Shared Rabbit Connection: SimpleConnection@631ca718 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 49982], acknowledgeMode=AUTO local queue size=0

但不幸的是，当我使用错误的凭据连接 Web 客户端时，我收到无限错误消息，可能是因为 AMQP 客户端正在重新发送有效负载。是否有某种方法可以引发异常并停止一次又一次地重新发送队列内容？

Answer 1

考虑捕获异常并将其包装到 AmqpRejectAndDontRequeueException 中:

/**
 * Exception for listener implementations used to indicate the
 * basic.reject will be sent with requeue=false in order to enable
 * features such as DLQ.
 * @author Gary Russell
 * @since 1.0.1
 *
 */
@SuppressWarnings("serial")
public class AmqpRejectAndDontRequeueException extends AmqpException {

另请参阅文档中有关此异常的一些用例，以及重试配置:https://docs.spring.io/spring-amqp/docs/current/reference/html/#async-listeners