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

Question

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

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 主题，其中终端节点对象采用queueUrl，如下所示：

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

借助 AWS Java SDK v2，您可以使用SunscribeRequest.builder方法将 SQS 队列订阅到 SNS 主题，其中终端节点对象采用queueArn。

下面是一个示例 Java 代码片段，我已经测试过该代码片段成功地将 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 队列，如下所示以及此处所述：

    {
      "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

希望这可以帮助。