AWS SDK 2 创建队列并订阅主题

Question

我正在使用 SNS / SQS 在 AWS 上工作，并且我正在尝试在 SQS 主题上创建 SNS 订阅。我正在使用 AWS Java SDK 2。我可以正确创建主题和队列，我面临的问题是在创建主题和队列之间的订阅时出现错误。以下是我的代码的 sn-p：

public SubscribeResponse subscribeQueue(String topicArn, String queueUrl){
    SubscribeRequest subscribeRequest = SubscribeRequest.builder()
            .topicArn(topicArn)
            .protocol("sqs")
            .endpoint(queueUrl)
            .returnSubscriptionArn(true).build();
    return getAmazonSNSClient().subscribe(subscribeRequest);
}

以下是我得到的错误：

{ "message": "Internal Server Error: software.amazon.awssdk.services.sns.model.InvalidParameterException: Invalid parameter: SQS endpoint ARN (Service: Sns, Status Code: 400, Request ID: dabaeeb3-e77a-5873-861e-b8d0a7bf5c7f)" }

谁能指出我做错了什么？

Answer 1

使用 AWS Java SDK v1，您可以使用方法 Topics.subscribeQueue 将 SQS 队列订阅到 SNS 主题，其中 endpoint 对象接收 queueUrl如下图：

Topics.subscribeQueue(sns, sqs, myTopicARN, queueURL);

使用 AWS Java SDK v2，您可以使用方法 SunscribeRequest.builder 将 SQS 队列订阅到 SNS 主题，其中 endpoint 对象接收 queueArn。

下面是一个示例 java 代码 sn-p，我已经测试它成功地将 SQS 队列订阅到 SNS 主题：

  public static void subscribeSQStoSNS(String queueUrl, String topicArn, SqsClient sqsClient, SnsClient snsClient) {


   /* with AWS Java SDK v1, you could subscribe an SQS queue to an SNS topic by
     * calling "Topics.subscribeQueue"()[1] where the endpoint object takes in queueUrl.
     * 
     * with AWS Java SDK v2, to subscribe an SQS queue to an SNS topic
     * one needs to use "SunscribeRequest.builder"() [2] where the endpoint object takes in queueArn.
     * 
     * [1] https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/sns/util/Topics.html
     * [2] https://sdk.amazonaws.com/java/api/2.0.0-preview-11/software/amazon/awssdk/services/sns/model/SubscribeRequest.Builder.html
     *
     */

  // Step 1: We call "get-queue-attributes" API to retrieve the SQS queue Arn

  GetQueueAttributesRequest QueueAttributesRequest = GetQueueAttributesRequest.builder()
            .queueUrl(queueUrl)
            .attributeNamesWithStrings("All")
            .build();

   GetQueueAttributesResponse QueueAttributesResult = sqsClient.getQueueAttributes(QueueAttributesRequest);
    Map<String, String> sqsAttributeMap = QueueAttributesResult.attributesAsStrings();

  System.out.println("\n\n"+ QueueAttributesRequest);
    String queueArn = sqsAttributeMap.get("QueueArn");
    //System.out.println("\n\n"+ queueArn);


  // Step 2: We call "SubscribeRequest.builder" to subscribe the SQS queue to the SNS topic

   SubscribeRequest Qrequest = SubscribeRequest.builder()
            .protocol("sqs")
            .endpoint(queueArn)
            .returnSubscriptionArn(true)
            .topicArn(topicArn)
            .build();

   SubscribeResponse Qresponse = snsClient.subscribe(Qrequest);
    System.out.println("\n\nCreated Subscription ARN: " + Qresponse.subscriptionArn()+ " " + " and StatusCode : " + Qresponse.sdkHttpResponse().statusCode());}

注意：

1. "Topics.subscribeQueue"() 方法会自动为订阅创建一个策略，以允许主题向队列发送消息。

2. "SunscribeRequest.builder"() 方法不会自动为订阅添加策略（即 SQS 队列），因此以允许 SNS 主题向队列发送消息。因此，请确保您已授予 SNS 主题发布到 SQS 队列的权限，方法是将以下示例策略添加到 SQS 队列，如下所示以及here 所述：

    {
      "Statement": [{
        "Effect":"Allow",
        "Principal":"*",
        "Action":"sqs:SendMessage",
        "Resource":"arn:aws:sqs:us-east-2:123456789012:MyQueue",
        "Condition":{
          "ArnEquals":{
            "aws:SourceArn":"arn:aws:sns:us-east-2:123456789012:MyTopic"
          }
        }
      }]
    }

您可以在 github 中找到完整的代码：https://gist.github.com/syumaK/79c946ed4e618c1f871f50ecb1819a0f

希望这会有所帮助。