使用JMS为带有SQS的nack定制接收超时

Question

否定确认的行为是将接收消息的可见性超时更改为0。其中，在为JMS创建SQS工厂时，NACK_TIMEOUT的值是不可配置的。

https://github.com/awslabs/amazon-sqs-java-messaging-lib/blob/master/src/main/java/com/amazon/sqs/javamessaging/acknowledge/NegativeAcknowledger.java#L99

当收到一条消息，并且处理失败（Listener方法抛出错误）时，该消息立即再次被接收。在大多数情况下，可以在一定的延迟时间内处理消息。

是否可以将其配置为不更改可见性超时，从而遵守队列的默认接收超时配置？

Answer 1

那些根本不想发送ChangeMessageViality请求（重置当前可见性计时器）并且只想依赖默认队列可见性超时的人可以使用以下在lombok帮助下编写的Amazon SQS包装器：

java prettyprint-override">public AmazonSQS amazonSqs(AWSCredentialsProvider credentialsProvider) {
    AmazonSQS amazonSQS = AmazonSQSClientBuilder.standard()
            .withRegion(region)
            (...)
            .build();

    return new IgnoringChangeVisibilityAmazonSqs(amazonSQS);
}

@Slf4j
@RequiredArgsConstructor
public class IgnoringChangeVisibilityAmazonSqs implements AmazonSQS {

    private interface ChangeVisibilityOperations {
        ChangeMessageVisibilityResult changeMessageVisibility(ChangeMessageVisibilityRequest changeMessageVisibilityRequest);
        ChangeMessageVisibilityResult changeMessageVisibility(String queueUrl, String receiptHandle, Integer visibilityTimeout);
        ChangeMessageVisibilityBatchResult changeMessageVisibilityBatch(ChangeMessageVisibilityBatchRequest request);
        ChangeMessageVisibilityBatchResult changeMessageVisibilityBatch(String queueUrl, List<ChangeMessageVisibilityBatchRequestEntry> entries);
    }

    @Delegate(excludes = ChangeVisibilityOperations.class)
    private final AmazonSQS amazonSqs;

    @Override
    public ChangeMessageVisibilityResult changeMessageVisibility(ChangeMessageVisibilityRequest changeMessageVisibilityRequest) {
        log.info("Ignoring ChangeMessageVisibilityRequest");
        return new ChangeMessageVisibilityResult();
    }

    @Override
    public ChangeMessageVisibilityResult changeMessageVisibility(String queueUrl, String receiptHandle, Integer visibilityTimeout) {
        log.info("Ignoring ChangeMessageVisibilityRequest");
        return new ChangeMessageVisibilityResult();
    }

    @Override
    public ChangeMessageVisibilityBatchResult changeMessageVisibilityBatch(ChangeMessageVisibilityBatchRequest request) {
        log.info("Ignoring ChangeMessageVisibilityBatchRequest");
        return changeMessageVisibilityBatch(request.getQueueUrl(), request.getEntries());
    }

    @Override
    public ChangeMessageVisibilityBatchResult changeMessageVisibilityBatch(String queueUrl, List<ChangeMessageVisibilityBatchRequestEntry> entries) {
        log.info("Ignoring ChangeMessageVisibilityBatchRequest");
        List<ChangeMessageVisibilityBatchResultEntry> results = entries.stream().map(request ->
                new ChangeMessageVisibilityBatchResultEntry().withId(request.getId()))
                .collect(Collectors.toList());
        return new ChangeMessageVisibilityBatchResult().withSuccessful(results);
    }
}

Answer 2

显然有一种方法有点笨拙，但我没有找到更好的解决方法，它的想法是，在创建客户端时，可以附加自己的请求处理程序：

    return AmazonSQSAsyncClientBuilder.standard()
        .withRequestHandlers(new RequestHandler2() {
            @Override
            public AmazonWebServiceRequest beforeExecution(AmazonWebServiceRequest request) {
                if (request instanceof ChangeMessageVisibilityBatchRequest) {
                    ((ChangeMessageVisibilityBatchRequest) request).getEntries().forEach(changeMessageVisibilityBatchRequestEntry -> {
                        changeMessageVisibilityBatchRequestEntry.setVisibilityTimeout(15);
                    });
                    log.debug("Changed visibility timeout for request {}", request);
                }
                return request;
            }
        })
    ;