java - 在 Spring AWS Cloud Framework 中设置 SNS WaitTime

Question

我正在使用 Spring AWS Cloud Framework轮询队列上的 S3 事件通知。我正在使用 QueueMessagingTemplate 来执行此操作。我希望能够设置最大消息数和轮询等待时间，请参阅:http://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html .

import com.amazonaws.services.s3.event.S3EventNotification;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.aws.messaging.core.QueueMessagingTemplate;

public class MyQueue {

    private QueueMessagingTemplate queueMsgTemplate;

    @Autowired
    public MyQueue(QueueMessagingTemplate queueMsgTemplate) {
        this.queueMsgTemplate = queueMsgTemplate;

    }

    @Override
    public S3EventNotification poll() {
        S3EventNotification s3Event = queueMsgTemplate
            .receiveAndConvert("myQueueName", S3EventNotification.class);
    }
}

上下文

@Bean
public AWSCredentialsProviderChain awsCredentialsProviderChain() {
    return new AWSCredentialsProviderChain(
            new DefaultAWSCredentialsProviderChain());
}

@Bean
public ClientConfiguration clientConfiguration() {
    return new ClientConfiguration();
}

@Bean
public AmazonSQS sqsClient(ClientConfiguration clientConfiguration,// UserData userData,
                           AWSCredentialsProviderChain credentialsProvider) {
    AmazonSQSClient amazonSQSClient = new AmazonSQSClient(credentialsProvider, clientConfiguration);
    amazonSQSClient.setEndpoint("http://localhost:9324");
    return amazonSQSClient;
}

@Bean
public QueueMessagingTemplate queueMessagingTemplate(AmazonSQS sqsClient) {
    return new QueueMessagingTemplate(sqsClient);
}

知道如何配置这些吗？谢谢

Answer 1

QueueMessagingTemplate.receiveAndConvert() 基于 QueueMessageChannel.receive() 方法，您可以在其中找到所需的代码:

@Override
public Message<String> receive() {
    return this.receive(0);
}

@Override
public Message<String> receive(long timeout) {
    ReceiveMessageResult receiveMessageResult = this.amazonSqs.receiveMessage(
            new ReceiveMessageRequest(this.queueUrl).
                    withMaxNumberOfMessages(1).
                    withWaitTimeSeconds(Long.valueOf(timeout).intValue()).
                    withAttributeNames(ATTRIBUTE_NAMES).
                    withMessageAttributeNames(MESSAGE_ATTRIBUTE_NAMES));
    if (receiveMessageResult.getMessages().isEmpty()) {
        return null;
    }
    com.amazonaws.services.sqs.model.Message amazonMessage = receiveMessageResult.getMessages().get(0);
    Message<String> message = createMessage(amazonMessage);
    this.amazonSqs.deleteMessage(new DeleteMessageRequest(this.queueUrl, amazonMessage.getReceiptHandle()));
    return message;
}

因此，如您所见，withMaxNumberOfMessages(1) 被硬编码为 1。这是正确的，因为 receive() 只能轮询一条消息。 withWaitTimeSeconds(Long.valueOf(timeout).intValue()) 完全等于提供的 timeout。呃，在 receiveAndConvert() 的情况下你不能修改它。

考虑使用 QueueMessagingTemplate 中的 QueueMessageChannel.receive(long timeout) 和 messageConverter。就像它是在:

public <T> T receiveAndConvert(QueueMessageChannel destination, Class<T> targetClass) throws MessagingException {
    Message<?> message = destination.receive();
    if (message != null) {
        return (T) getMessageConverter().fromMessage(message, targetClass);
    } else {
        return null;
    }
}

您可以通过代码访问适当的QueueMessageChannel:

String physicalResourceId = this.destinationResolver.resolveDestination(destination);
new QueueMessageChannel(this.amazonSqs, physicalResourceId);