AWS 集成弹簧：延长可见性超时答案

【问题标题】：Aws integration spring: Extend Visibility TimeoutAWS 集成弹簧：延长可见性超时
【发布时间】：2016-11-16 06:28:38
【问题描述】：

是否可以延长正在传输的消息的可见性时间。

见：

http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/AboutVT.html。

部分：更改消息的可见性超时。

http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/sqs/AmazonSQSClient.html#changeMessageVisibility-com.amazonaws.services.sqs.model.ChangeMessageVisibilityRequest-

总之，我希望能够为正在传输的给定消息延长第一次设置的可见性超时。

例如，如果 15 秒过去了，我想将超时再延长 20 秒。上面的 java 文档中有更好的例子。

根据我在上述链接中的理解，您可以在亚马逊方面执行此操作。

以下是我目前的设置；

  SqsMessageDrivenChannelAdapter adapter =
  new SqsMessageDrivenChannelAdapter(queue);
  adapter.setMessageDeletionPolicy(SqsMessageDeletionPolicy.ON_SUCCESS);
  adapter.setMaxNumberOfMessages(1);
  adapter.setSendTimeout(2000);
  adapter.setVisibilityTimeout(200);
  adapter.setWaitTimeOut(20);

是否可以延长此超时时间？

【问题讨论】：

抱歉，您的问题不清楚。我们只公开目标ReceiveMessageRequest 中存在的那些属性。我只是不明白你还想看到什么。
@ArtemBilan 我希望现在已经清楚了。

标签： spring spring-integration aws-sdk spring-cloud

【解决方案1】：

好的。看来我明白你的意思了。

我们可以使用 API 更改特定消息的可见性：

AmazonSQS.changeMessageVisibility(String queueUrl, String receiptHandle, Integer visibilityTimeout)

为此，在下游流程中，您必须访问（注入）AmazonSQS bean 并从 Message 中提取特殊标头：

@Autowired
AmazonSQS amazonSqs;

@Autowired
ResourceIdResolver resourceIdResolver;
...


MessageHeaders headers = message.getHeaders();

DestinationResolver destinationResolver = new DynamicQueueUrlDestinationResolver(this.amazonSqs, this.resourceIdResolver);

String queueUrl = destinationResolver.resolveDestination(headers.get(AwsHeaders.QUEUE));

String receiptHandle = headers.get(AwsHeaders.RECEIPT_HANDLE);

amazonSqs.changeMessageVisibility(queueUrl, receiptHandle, YOUR_DESIRED_VISIBILITY_TIMEOUT);

但是，我同意我们应该提供一些关于这个问题的东西作为开箱即用的功能。这甚至可能类似于 QueueMessageAcknowledgment 作为新标头。或者甚至只是一个changeMessageVisibility这个方法。

请为Spring Cloud AWS 项目提出 GH 问题，并附上此 SO 主题的链接。

【讨论】：

看起来我们已经有了一个：github.com/spring-cloud/spring-cloud-aws/issues/92

【解决方案2】：

Spring Cloud AWS 从版本 2.0 开始支持这一点。在 SQS 侦听器方法中注入 Visiblity 参数就可以了：

  @SqsListener(value = "my-sqs-queue")
  void onMessageReceived(@Payload String payload, Visibility visibility) {
    ...
    var extension = visibility.extend(20);
    ...
  }

注意，extend 将异步工作并返回一个 Future。因此，如果您想进一步确定处理过程，消息的可见性确实在 AWS 方面得到了扩展，可以使用 extension.get() 阻止 Future 或使用 extension.isDone() 查询 Future

【讨论】：