公有云提倡使用分布式应用，利用大量的服务器资源来运行多个系统，从那些大型的企业应用到简单的部门级服务。然而对于管理员来说，他们该如何轻松地将数据移动到合适的服务器上呢？其中一种方式就是使用消息总线。如果你想要在从多个数据源中以实时的方式来处理数据，那么Amazon Kinesis会是不错的选择。

Amazon Kinesis 是AWS提供的一个托管消息服务，它能够实现高性能的实时数据处理。该服务设计用来从大量数据源中接受消息，然后将它们分布到各个应用系统当中。Kinesis是基于 Apache Kafka构建的服务，后者主要提供消息发布与订阅服务。 

配置AmazonKinesis

在设置一个Amazon Kinesis发布与订阅平台的时候，首先需要定义一个Kinesis流。它通常是在AWS Management Console完成的。

Kinesis流是一系列资源，它能够对消息进行接收、存储并转换。海量数据流可以在多个shard中进行分割，这与使用多个服务器进行集群处理相类似。对shard数量的需求取决于消息的平均大小，以及记录写入的比例以及应用的数量。AWS Management Console中自带一个工具，可以帮助管理员设定所需shard的数量，从而满足数据处理的需求。

与其他AWS资源一样，你需要设置访问控制。Kinesis权限使得你可以指定用户与角色，比如将消息放入队列，获取队列的状态信息以及从队列中读取数据。

生成者应用将被服务向Kinesis队列写入消息的权限。一条消息包含了将要提交的值或者有效负载，比如一个包含键值对的JSON结构，还有一个shard键。一个shard每秒最多可以接收1000条消息；shard键最多可以有256个字符的长度。Kinesis API提供了一个PutRecord函数，它主要控制每次向队列中提交一条消息；而PutRecords函数则可以提交多条消息。 

应用会调用GetRecords API函数。通常来说，目标应用会持续循环式地运行这些函数。每条记录最多支持50KB的大小。在一个shard中，系统每秒可以应对2MB的数据。如果你需要额外的吞吐能力，你可以向数据流中添加更多的shard来实现。GetRecords函数还支持一个LIMIT参数来指定记录的最大值。管理员可以使用这个功能来调整应用可接受的数据量，特别是在消息队列的写入高峰期。

AmazonKinesis的缺点

当进行实时数据处理的时候，Kinesis会有一些限制。服务对消息最多只保留24小时。而Kafka有所不同，它保留消息的时间会长很多。IT团队需要为应用分配好资源，从而能够在24小时之内完成所有消息的读取。

AWS CloudWatch 可以帮助监控消息队列的负载以及应用的吞吐量。AWS Elastic Load Balance或Auto Scaling可以确保有足够的计算资源来处理消息数据流。

如果你超过了Kinesis的限制，你将收到ProvisionedThroughtputExceed报错；然后消息将被拒绝。如果在读操作阶段超出限制，你也会收到ProvisionedThroughtputExceed报错。如果想增加容量，你可以向数据流中添加更多的shard。

再添加shard的时候，你可以将现有的shard进行拆分。每次拆分只需要几秒钟时间，但你每次只能拆分一个shard。所以，当你上百个shard的时候，就会非常麻烦。

开发与集成

Kinesis提供了REST API，因此你几乎可以使用任意一种编程语言来对消息队列进行写入与读取。AWS软件开发工具同样提供了针对特定编程语言的Kinesis函数。

AWS还提供了几个连接器，用来进行与其他服务的集成，比如DynamoDB、Redshift、Simple Storage Service 以及ElasticSearch等等。Kinesis会根据shard使用时间来收取费用，价格为每小时0.015美元，它还可以按照PUT操作的数量来收费，价格为每100万个PUT 0.028美元。