来自：http://www.ibm.com/developerworks/cn/java/j-javadev2-17/
消息传送队列在一系列软件架构和域中是常见的，包括金融系统、医疗保健和旅游业。然而面向消息的中间件（MOM）— 分布式系统的主导消息传送范例 — 需要特别安装和维护一个队列系统。本月我对这种劳动密集型消息传送引入了一种基于云计算的替代方案：Amazon 的 Simple Queue Service (SQS)。
就像在 Google App Engine 或 Amazon Elastic Beanstalk（参见 参考资料）上托管 web 应用程序很合乎常理一样，利用一个云消息传送系统也合乎常理。不管怎样，您都能够将更多的时间用于编写应用程序，而非安装和维护其底层基础架构。
在本文中，您将了解 Amazon SQS 如何减轻安装和维护一个消息队列系统的负担。您还有机会亲手创建 SQS 消息队列，然后删除和检索上面的消息。最后，我要向您展示在我添加消息到 Magnus 时会发生什么，这是我在上个月的 Amazon Elastic Beanstalk 简介中使用的移动 web 应用程序。
面向消息的中间件 或 MOM 是一个描述通过消息队列通信的松耦合系统的术语。系统组件没有进行紧耦合（例如通过编译时依赖项），而是被分布在整个网络中。这个分布式效果以消息队列 为通信媒介，能够让消息系统进行扩展。
传统上，架构师决定了面向消息的系统中哪些组件将互相通信。尽管所有通信通过消息传递发生，消息本身通常是一个通用的跨平台格式。消息可以是简单的字符串或甚至使用 XML 或 JSON 编码的文档。
由于 MOM 架构对组件去耦且支持它们之间的跨平台通信，单个原件可以是异构的。也就是说，分布式架构中的组件可通过不同的语言编写，比如 Java 语言、C# 和 Ruby。组件也可以存在于不同的平台上，比如 UNIX® 和 Windows®。更重要的是，MOMs 使系统集成更加容易。作为中间件，MOMs 可以连接旧有系统以及新系统。这是因为组件之间的 API 仅仅是一条消息，可以是从 XML 文档到序列化对象再到简单 String 的任意内容。
MOM 系统中的消息队列是 web 的管道：它们连接各种系统组件以允许消息在它们之间自由流动。事实证明，GAE 是面向消息中间件系统的一个很好的例子。
如同任何好的 MOM，Google App Engine 使用消息队列分离系统流程。特别地，GAE 队列使我们可以通过 Web 请求卸载长期运行的流程。您可以使用 GAE 将指向 servlets 或 JSPs 的 URLs 转储到消息队列上，然后由 GAE 服务选取和处理。Servlets 被根据 web 应用程序的主逻辑顺序异步调用。（参见 参考资料 进一步了解 GAE。）
但是排列长期运行的流程来管理主流程的持续时间不仅仅是一个 GAE 工作。这个类似 MOM 的特性随 Heroku 等其他 PaaS 实现一同提供。然而使用 Amazon SQS，您可以轻松在任何 web 应用中进行，而不管平台是什么。
如果您在 JMS 中使用了消息队列，Amazon SQS 提供大量您应当很熟悉的功能。
Amazon SQS：
- 允许多个进程从同一队列同时进行读取操作。它还在处理期间锁定消息，确保一条消息仅由一个读取器处理，即使多个进程在从单一队列进行读取操作。
- 利用 Amazon 的大量冗余架构在并发访问时提供极高的可用性。它还保证消息的交付（至少一次）。
- 需要您仅根据使用量进行支付。对于 Amazon SQS，这表示每条消息您支付 $0.000001。AWS 当前提供一个免费的层级，其中每个月前 100,000 条消息是免费的。记住有按千兆字节定价的带宽费用，这对所有 AWS 产品都是通用的。
SQS 入门就如同 AWS 中一切别的东西一样简单。如果您还没有一个 AWS 帐户，首先 创建一个。其次，启用 Amazon SQS。最后，使用 AWS 接口 Java SDK 发布和阅读基于云的消息！（下面将详细介绍如何实际编写 它们。）
与 Amazon SQS 名称相一致，读写到队列背后的逻辑本身很简单。首先，使用有效访问密匙和机密建立 AWS 连接，如清单 1 所示：
AmazonSQS sqs = new AmazonSQSClient(new BasicAWSCredentials(AWS_KEY, AWS_SECRET)); |
下一步，您需要一个队列。在 AWS API 中，对 createQueue 的调用，如清单 2 所示，不一定每次创建一个新队列 。如果队列已经存在，返回其句柄。在 SQS 中，队列仅仅是 URLs；因此，队列处理也只是一个 URL。注意在 AWS SDK API 中，Queue URL 是一个String 类型，而不是 Java URL 类型。
String url = sqs.createQueue(new CreateQueueRequest("a_queue")).getQueueUrl();
|
有了队列之后，您可以向其写入一条消息。SQS 的消息格式类似于 SimpleDB 的（参见 参考资料），在于消息是 Strings。不过要记住，一条 String 是可以轻松结构化的，因此易于解析，即使其格式为有效的 JSON 或 XML。
sqs.sendMessage(new SendMessageRequest(url, "It's a wonderful life!")); |
消息长度是有限的。默认情况下，一条消息不能超过 8KB。如果您需要使用较长的消息，总是可以将它们分成小块，使用序列 IDs 识别单个部分。然后可以在接收方重新组合消息。
就这么简单 — 将一条消息放到 SQS 队列仅需要那三行代码。
您可能会注意到 AWS SDK 中的一个熟悉模式，特别是如果您阅读了我的 SimpleDB 简介（参见 参考资料）。由于 AWS 中的一切就是一个 web 服务，所有通信通过 HTTP 发生。因此，API 通过与 Request 类似的对象模拟逻辑请求，比如 SendMessageRequest 或CreateQueueRequest。在两种情况下，名称描述对象的意图。
另外要注意的是，放在 SQS 上的消息是持久的：在删除之前它们一直在那里。（如果您不删除它们，消息最终不会消失；自动过期的默认值为 4 天。）当获取消息来阅读时，Amazon SQS 采用简单的锁定策略 — 用于阅读事件，消息在一个时期内不可用于其他并发读取进程，这被称为消息的可视性超时。该值被默认设置为 30 秒，不过您可以随需自由改变持续时间。
位于 Amazon 架构的消息的持久性是可靠的。如同 SimpleDB 甚至 S3，AWS 中的组件是大量冗余的。如果您的读取器进程在消息处理期间意外终止，很有可能消息还在。而且，如果 AWS 网络中的一些资产也决定被销毁，您可以确信您的任务关键型消息不会被丢失 — 它们将继续存在于任意数量的其他机器上。最后，所有其他 AWS 产品通常都是这样，您可以按地区设置您的消息基础架构的物理位置：美国，欧盟等。
向一个 SQS 队列写一条消息需要三行代码。读取消息稍微多一点。事实上，前两行是一样的，鉴于您需要一个 AWS 连接和同一队列的句柄。Amazon SQS 不提供任何回调功能或消息到达的预先通知。您必须定期轮询一个 SQS 队列，看看它是否有什么可提供的。因此，读取一个 SQS 队列需要这些额外的代码行。
在实现轮询策略时要稍微注意一下：您必须检查并确保在处理一条消息之前实际接收了一条有效消息。如果您不检查，最终一定会看到这个可恶的 NullPointerException。
例如，假设我获取一个有效的 AWS 连接和包含消息的队列的句柄，我可以如清单 4 所示检索消息：
while (true) {
List<Message> msgs = sqs.receiveMessage(
new ReceiveMessageRequest(url).withMaxNumberOfMessages(1)).getMessages();
if (msgs.size() > 0) {
Message message = msgs.get(0);
System.out.println("The message is " + message.getBody());
sqs.deleteMessage(new DeleteMessageRequest(url, message.getReceiptHandle()));
} else {
System.out.println("nothing found, trying again in 30 seconds");
Thread.sleep(3000);
}
}
|
在清单 4 中，对 sqs 的引用是一个 AmazonSQS 类，如 清单 1 所示。该对象提供一个 receiveMessage 方法，该方法接受一个ReceiveMessageRequest。可以配置 ReceiveMessageRequests 来请求队列中的一组消息。在我的例子中，我将其配置为一次仅获取一条消息。不管我请求多少条消息，receiveMessage 方法都返回 Message 类型的一个 List。
正如我前面提到的，SQS 读取是通过轮询方式完成的；而且 receiveMessage 方法是非阻塞的。因此，我必须检查确保相应的List（msgs）确实包含什么东西。如果从队列上未检索到任何东西，对 ReceiveMessageRequest 上 getMessages 的调用会返回一个空的 List，而非 null。
只要我检索了有效消息，我就可以通过 getBody 调用获取其负载或主体。请记住，在有了有效消息句柄之后，SQS 会锁定它。默认情况下我有 30 秒的时间对消息做一些事情。如果我希望永久地从处理中移除消息，我必须删除它。因此，我发出一个 deleteMessage调用，该调用需要一个 DeleteMessageRequest。
一个 Message 实例的特点是它的接收句柄，如同 id。句柄不直接与消息相关，但是与读取它的事件 相关。读取过多次的一条消息（比如如果它没有被删除或则读取过程失败）可以有多个不同的接收句柄，最终当您希望删除一条消息时，您必须通过getReceiptHandle 调用提供其接收句柄。
我没有不断检查看我的队列是否有消息，我提供了一个休眠函数，如没有检索到消息就等待 30 秒。显然在有些情况下，休眠可能不是一个好主意，或者一个较长时间的停顿会很合适。
有了这几行代码，我就将 Amazon SQS 介绍得差不多了。虽然 AWS SDK 提供大量其他功能和特性，目前为止的代码是您读写消息到 SQS 队列所需的一切。
现在我们来看看真正使用它时会发生什么。
上个月，我创建了一个简单的移动 web 应用程序 Magnus，我使用它来展示 Amazon Elastic Beanstalk 的一些特性（参见 参考资料）。Magnus 有一个不错的功能，可以存储接收自帐户持有人的移动设备的位置信息 — 就是很多人希望提供且其他人希望使用的那种信息。
捕捉某些人的行踪是好的，但是人们真正喜欢的是图形（以及闪亮的圆角按钮）。当您有大量数据要移动时，从处理角度来看图形化和分析会很昂贵。（Hadoop 适用于任何人？）行之有效的提取、转换和加载 或 ETL 技术是管理这个的一种方式。ETL 是包含大量东西的一个相当大的术语。（人们围绕这个缩写打造职业生涯，公司围绕这个建立业务！）在本例中，ETL 仅表示我要分析一些 MongoDB 数据并基于该数据创建新文档。
谈及数据分析时，对于我们对数据的要求以及我们可以提供的答案有多种可能性。Magnus web 应用程序呈现了这种可能性的一小部分：它提取并呈现与地理坐标、时间和用户帐户相关的数据。从技术上讲，Magnus 关注位置经纬度、用户帐户 ID、时间戳和这些特定数据之间的关系。
Magnus 可以通过图形表示该数据，根据地理区域展示用户帐户（可能是一幅地图，使用标记在给定时间定位帐户持有人）。或者它可以展示如何跨给定区域（另一幅地图）移动一个帐户持有人/用户。提供这种信息需要一个离线的 ETL 式流程。从处理角度来看，在生成数据时实时提供该数据会很昂贵。因此将这些分析看作是近实时的。
为在 Magnus 中使用 Amazon SQS，我需要做一些初步设置。首先，我需要一种获取 AWS 凭据的方式。我喜欢 Play（参见 参考资料），因此我要将它作为我的应用程序开发框架。要获取凭据，我可以使用 Play 的 application.conf 文件，即一个自动读取的属性文件。
清单 5. 添加 AWS 配置数据到 Play 的 application.conf
#AWS configuration aws_access_key_id=1S..........MR2 aws_secret_access_key=S3.........ZM |
在定义了属性之后，我可以通过对 Play 的 Play 对象的一个调用来轻松获取它们，如清单 6 所示：
public class Application extends Controller {
private static final String AWS_KEY =
Play.configuration.get("aws_access_key_id").toString();
private static final String AWS_SECRET =
Play.configuration.get("aws_secret_access_key").toString();
//....
}
|
定义了该管道之后，我就可以开始正事了。清单 7 中的代码类似于我上个月的 Amazon Elastic Beanstalk 简介中使用的一个代码段。在本例中，我仅使用一些代码更新了 saveLocation，以将一个简单的 JSON 文档放到名为 “locations_queue” 的一个队列上。JSON 基本上是这样的：{"id":"4d6baeb52a54f1000001"}。已存位置的 ID 提供给了消息接收人，供其进行查询和分析。
清单 7. 将消息放到 SQS 上的一个 saveLocation 方法
public static void saveLocation(String id, JsonObject body) throws Exception {
String eventname = body.getAsJsonPrimitive("name").getAsString();
double latitude = body.getAsJsonPrimitive("latitude").getAsDouble();
double longitude = body.getAsJsonPrimitive("longitude").getAsDouble();
String when = body.getAsJsonPrimitive("timestamp").getAsString();
SimpleDateFormat formatter =
new SimpleDateFormat("dd-MM-yyyy HH:mm");
Date dt = formatter.parse(when);
ObjectId oid = new Location(id, dt, latitude, longitude).save();
AmazonSQS sqs = new AmazonSQSClient(new BasicAWSCredentials(AWS_KEY, AWS_SECRET));
Map mp = new HashMap<String, String>();
mp.put("id", oid.toString());
String url = sqs.createQueue(new CreateQueueRequest("locations_queue")).getQueueUrl();
sqs.sendMessage(new SendMessageRequest(url, new Gson().toJson(mp)));
renderJSON(getSuccessMessage());
}
|
在将该消息放到 SQS 队列上之后，我需要从队列中选取它们并执行一些处理。不知您是否记得，MOM 的优势之一是它允许异构架构。为此，SQS 读取器方可使用 Java 之外的语言编写，甚至在另一个平台上运行！
因为我基本上可以用我喜欢的任何语言执行分析处理，我要用 Ruby 来执行 — 来赢得时尚年轻人的一些街头口碑。
在清单 8 中，我借助了 right_aws Ruby gem 来帮助我处理 SQS。在很多方面，您可以将 gem 看作是 jar 文件。right_aws 库很像 Amazon 的 SDK for Java，不过没那么冗长且使用起来更简单直观。
require "right_aws"
#...
sqs = RightAws::SqsGen2.new(aws_access_key_id, aws_secret_access_key)
queue = sqs.queue('locations_queue')
|
如您所见，清单 8 中的两行相关代码建立 AWS 连接并获取我的 'locations_queue' 队列的句柄。
接下来，我建立了一个轮询机制，如清单 9 所示。对 @queue 的引用是清单 8 中的同一 queue 变量。不过在本例中，它被定义为类的一部分。因此在清单 9 中，我使用 Ruby 的 @ 语法直接引用一个实例变量。
def process_messages()
while true
msg = @queue.pop
if !msg.nil?
handle_message(msg) # impl of which does neat stuff
msg.delete
else
sleep 10
end
end
end
|
在将消息传递给 handle_message 方法之后，我可以删除它。如果未找到任何消息，主线程休眠 10 秒钟。!msg.nil? 一行类似于 Java 代码中的 msg != null。但是在 Ruby 中，甚至 null 也是一个对象。询问一个对象是否是 nil 类型（通过 nil? 方法调用）会返回一个布尔值。
因为 AWS 是一个 web 服务产品，它为众多平台库所访问和利用。在 Magnus 中，您可以看到所产生的灵活性：我能够使用 Java 代码将消息推送到一个 SQS 队列，然后使用一个小 Ruby 程序将它们剥离下来。运用查询的架构的美妙之处在于组件的隐式解耦。
就像在 GAE 或 Amazon 的 Elastic Beanstalk 上托管一个 web 应用程序通常很合理一样，利用云消息传送系统也很合理。Amazon 的 SQS 减轻了安装和维护队列系统的负担。您仅需创建一个队列，然后在其上删除和检索消息。将其余工作留给 Amazon。