Amazon Simple Queue Service
开发人员指南
AWS 服务或AWS文档中描述的功能,可能因地区/位置而异。点 击 Getting Started with Amazon AWS to see specific differences applicable to the China (Beijing) Region.

标准队列

Amazon SQS 提供了标准 作为默认队列类型。标准队列使您能够每秒处理近乎无限数量的事务。标准队列支持至少一次消息传送。但是,由于存在允许近乎无限吞吐量的高度分布式架构,偶尔会有一条消息的多个副本不按顺序传送。标准队列提供了最大努力排序,保证了消息大致按其发送的顺序进行传送。

您可以在很多情况下使用标准消息队列 (只要应用程序能处理多次到达和不按顺序到达的消息),例如:

  • 将实时用户请求从密集的后台工作中解耦 – 让用户在调整媒体大小或对媒体编码时上传媒体。

  • 将任务分配给多个工作程序节点 – 处理大量信用卡验证请求。

  • 将消息分批以便进一步处理 – 计划要添加到数据库的多个条目。

有关使用标准队列的最佳实践,请参阅一般建议

消息排序

标准队列会尽量保持消息顺序,但可能有一条消息的多个副本可能不按顺序传送。如果您的系统要求保留该顺序,我们建议使用 FIFO (先进先出)队列,或者在每条消息中添加排序信息,以便您在收到消息后对其重新排序。

至少一次传递

Amazon SQS 会在多台服务器上存储消息的副本,以实现冗余和高可用性。在极少数情况下,当您接收或删除消息时,存储消息副本的某台服务器可能不可用。

如果出现这种情况,则该不可用服务器上的消息副本将不会被删除,并且您在接收消息时可能会再次获得该消息副本。将应用程序设计为幂等 应用程序 (多次处理同一消息时,它们不应受到不利影响)。

通过短轮询使用消息

从队列中使用消息的行为取决于您使用的是短 (标准) 轮询 (默认行为) 还是长轮询。有关长轮询的更多信息,请参阅“Amazon SQS 长轮询”。

当通过短轮询从队列中使用消息时,Amazon SQS 会对服务器的一个子集 (基于加权随机分布) 进行采样,并且仅从这些服务器返回消息。因此,特定接收请求可能不会返回您的所有消息。不过,如果您队列中的消息数比较少 (少于 1000 条),则一个特定请求可能不会返回您的任何消息,而后续请求则会返回您的任何消息。如果您继续从您的队列中使用消息,则 Amazon SQS 会对所有服务器进行采样,然后您将收到您的所有消息。

下图显示了您的系统组件之一提出接收请求后返回的消息的短轮询行为。Amazon SQS 对若干服务器 (显示为灰色) 进行采样并从这些服务器返回消息 (消息 A、C、D 和 B)。系统不会为此特定请求返回消息 E,但将为后续请求返回该消息。

 使用短 (标准) 轮询的消息采样

标准 队列入门指南

以下示例 Java 代码将创建队列并发送、接收和删除消息。

Copy
/* * Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved. * * Licensed under the Apache License, Version 2.0 (the "License"). * You may not use this file except in compliance with the License. * A copy of the License is located at * * http://aws.amazon.com/apache2.0 * * or in the "license" file accompanying this file. This file is distributed * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either * express or implied. See the License for the specific language governing * permissions and limitations under the License. */ import java.util.List; import java.util.Map.Entry; import com.amazonaws.AmazonClientException; import com.amazonaws.AmazonServiceException; import com.amazonaws.auth.AWSCredentials; import com.amazonaws.auth.profile.ProfileCredentialsProvider; import com.amazonaws.regions.Region; import com.amazonaws.regions.Regions; import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.AmazonSQSClientBuilder; import com.amazonaws.services.sqs.model.CreateQueueRequest; import com.amazonaws.services.sqs.model.DeleteMessageRequest; import com.amazonaws.services.sqs.model.DeleteQueueRequest; import com.amazonaws.services.sqs.model.Message; import com.amazonaws.services.sqs.model.ReceiveMessageRequest; import com.amazonaws.services.sqs.model.SendMessageRequest; /** * This sample demonstrates how to make basic requests to Amazon SQS using the * AWS SDK for Java. * * Prerequisites: You must have a valid Amazon Web Services developer account, * and be signed up to use Amazon SQS. For more information about Amazon SQS, * see http://aws.amazon.com/sqs * * Fill in your AWS access credentials in the provided credentials file * template, and be sure to move the file to the default location * (~/.aws/credentials) where the sample code loads the credentials from. * * IMPORTANT: To avoid accidental leakage of your credentials, DO NOT * keep the credentials file in your source directory. */ public class SimpleQueueServiceSample { public static void main(String[] args) throws Exception { /* * The ProfileCredentialsProvider returns your [default] * credential profile by reading from the credentials file located at * (~/.aws/credentials). */ AWSCredentials credentials = null; try { credentials = new ProfileCredentialsProvider().getCredentials(); } catch (Exception e) { throw new AmazonClientException( "Cannot load the credentials from the credential profiles file. " + "Please make sure that your credentials file is at the correct " + "location (~/.aws/credentials), and is in valid format.", e); } AmazonSQS sqs = AmazonSQSClientBuilder.standard() .withRegion(Regions.US_WEST_2) .build(); System.out.println("==========================================="); System.out.println("Getting Started with Amazon SQS"); System.out.println("===========================================\n"); try { // Create a queue System.out.println("Creating a new SQS queue called MyQueue.\n"); CreateQueueRequest createQueueRequest = new CreateQueueRequest("MyQueue"); String myQueueUrl = sqs.createQueue(createQueueRequest).getQueueUrl(); // List queues System.out.println("Listing all queues in your account.\n"); for (String queueUrl : sqs.listQueues().getQueueUrls()) { System.out.println(" QueueUrl: " + queueUrl); } System.out.println(); // Send a message System.out.println("Sending a message to MyQueue.\n"); sqs.sendMessage(new SendMessageRequest(myQueueUrl, "This is my message text.")); // Receive messages System.out.println("Receiving messages from MyQueue.\n"); ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(myQueueUrl); List<Message> messages = sqs.receiveMessage(receiveMessageRequest).getMessages(); for (Message message : messages) { System.out.println(" Message"); System.out.println(" MessageId: " + message.getMessageId()); System.out.println(" ReceiptHandle: " + message.getReceiptHandle()); System.out.println(" MD5OfBody: " + message.getMD5OfBody()); System.out.println(" Body: " + message.getBody()); for (Entry<String, String> entry : message.getAttributes().entrySet()) { System.out.println(" Attribute"); System.out.println(" Name: " + entry.getKey()); System.out.println(" Value: " + entry.getValue()); } } System.out.println(); // Delete a message System.out.println("Deleting a message.\n"); String messageReceiptHandle = messages.get(0).getReceiptHandle(); sqs.deleteMessage(new DeleteMessageRequest(myQueueUrl, messageReceiptHandle)); // Delete a queue System.out.println("Deleting the test queue.\n"); sqs.deleteQueue(new DeleteQueueRequest(myQueueUrl)); } catch (AmazonServiceException ase) { System.out.println("Caught an AmazonServiceException, which means your request made it " + "to Amazon SQS, but was rejected with an error response for some reason."); System.out.println("Error Message: " + ase.getMessage()); System.out.println("HTTP Status Code: " + ase.getStatusCode()); System.out.println("AWS Error Code: " + ase.getErrorCode()); System.out.println("Error Type: " + ase.getErrorType()); System.out.println("Request ID: " + ase.getRequestId()); } catch (AmazonClientException ace) { System.out.println("Caught an AmazonClientException, which means the client encountered " + "a serious internal problem while trying to communicate with SQS, such as not " + "being able to access the network."); System.out.println("Error Message: " + ace.getMessage()); } } }